将RDD转换为DataFrame Spark Streaming时的ClassCastException

Question

嗨，伙计们，我有下一个问题。我正在使用Java的Apache Spark Streaming v1.6.0来获取来自IBMMQ的一些消息。我为MQ制作了自定义接收器，但我遇到的问题是我需要将RDD从JavaDStream转换为DataFrame。为此，我使用foreachRDD迭代JavaDStream，并定义了DataFrame的模式，但当我运行作业时，第一条消息会引发下一个异常：

Java语言lang.ClassCastException：组织。阿帕奇。火花rdd。BlockRDDPartition无法转换为组织。阿帕奇。火花rdd。组织上的ParallelCollectionPartition。阿帕奇。火花rdd。并行集合RDD。计算（ParallelCollectionRDD.scala:102）。阿帕奇。火花rdd。RDD。位于组织的computeOrReadCheckpoint（RDD.scala:306）。阿帕奇。火花rdd。RDD。org上的迭代器（RDD.scala:270）。阿帕奇。火花调度程序。结果任务。在组织上运行任务（ResultTask.scala:66）。阿帕奇。火花调度程序。任务在组织上运行（Task.scala:89）。阿帕奇。火花执行人。执行者$TaskRunner。在java上运行（Executor.scala:213）。util。同时发生的线程池执行器。java上的runWorker（ThreadPoolExecutor.java:1149）。util。同时发生的ThreadPoolExecutor$工作者。在java上运行（ThreadPoolExecutor.java:624）。lang.Thread。run（Thread.java:748）19/03/28 12:53:26警告TaskSetManager：在0.0阶段（TID 0，localhost）丢失任务0.0：java。lang.ClassCastException：组织。阿帕奇。火花rdd。BlockRDDPartition无法转换为组织。阿帕奇。火花rdd。组织上的ParallelCollectionPartition。阿帕奇。火花rdd。并行集合RDD。计算（ParallelCollectionRDD.scala:102）。阿帕奇。火花rdd。RDD。位于组织的computeOrReadCheckpoint（RDD.scala:306）。阿帕奇。火花rdd。RDD。org上的迭代器（RDD.scala:270）。阿帕奇。火花调度程序。结果任务。在组织上运行任务（ResultTask.scala:66）。阿帕奇。火花调度程序。任务在组织上运行（Task.scala:89）。阿帕奇。火花执行人。执行者$TaskRunner。在java上运行（Executor.scala:213）。util。同时发生的线程池执行器。java上的runWorker（ThreadPoolExecutor.java:1149）。util。同时发生的ThreadPoolExecutor$工作者。在java上运行（ThreadPoolExecutor.java:624）。lang.Thread。运行（Thread.java:748）

然后代码执行得很好。即使我在MQ中没有任何消息，也只是我运行作业时的第一条消息。

这是我的CustomMQReceiver

public CustomMQReceiver() {

        super(StorageLevel.MEMORY_ONLY_2());

    }

    @Override
    public void onStart() {

        new Thread() {
            @Override
            public void run() {
                try {
                    initConnection();
                    receive();
                } catch (JMSException ex) {
                    ex.printStackTrace();
                }
            }
        }.start();

    }

    @Override
    public void onStop() {

    }

    private void receive() {

        System.out.print("Started receiving messages from MQ");

        try {

            Message receivedMessage = null;

            while (!isStopped() && (receivedMessage = consumer.receiveNoWait()) != null) {

                String userInput = convertStreamToString(receivedMessage);
                System.out.println("Received data :'" + userInput + "'");
                store(userInput);
            }

            stop("No More Messages To read !");
            qCon.close();
            System.out.println("Queue Connection is Closed");

        } catch (Exception e) {
            e.printStackTrace();
            restart("Trying to connect again");
        } catch (Throwable t) {

            restart("Error receiving data", t);
        }

    }

    public void initConnection() throws JMSException {

        MQQueueConnectionFactory conFactory = new MQQueueConnectionFactory();
        conFactory.setHostName(HOST);
        conFactory.setPort(PORT);
        conFactory.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
        conFactory.setQueueManager(QMGR);
        conFactory.setChannel(CHANNEL);
        conFactory.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
        conFactory.setStringProperty(WMQConstants.USERID, APP_USER);
        conFactory.setStringProperty(WMQConstants.PASSWORD, APP_PASSWORD);

        qCon = (MQQueueConnection) conFactory.createConnection();
        MQQueueSession qSession = (MQQueueSession) qCon.createQueueSession(false, 1);
        MQQueue queue = (MQQueue) qSession.createQueue(QUEUE_NAME);
        consumer = (MQMessageConsumer) qSession.createConsumer(queue);
        qCon.start();

    }

    @Override
    public StorageLevel storageLevel() {
        return StorageLevel.MEMORY_ONLY_2();
    }

    private static String convertStreamToString(final Message jmsMsg) throws Exception {

        String stringMessage = "";
        JMSTextMessage msg = (JMSTextMessage) jmsMsg;
        stringMessage = msg.getText();

        return stringMessage;
    }

这是我的spark代码

SparkConf sparkConf = new SparkConf()
                    .setAppName("MQStreaming")
                    .set("spark.driver.allowMultipleContexts", "true")
                    .setMaster("local[*]");

            JavaSparkContext jsc = new JavaSparkContext(sparkConf);
            final SQLContext sqlContext = new SQLContext(jsc);
            JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(Long.parseLong(propertiesConf.getProperty("duration"))));

            JavaDStream<String> customReceiverStream = ssc.receiverStream(new CustomMQReceiver());

            customReceiverStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {

                @Override
                public void call(JavaRDD<String> rdd) throws Exception {

                    JavaRDD<Row> rddRow = rdd.map(new Function<String, Row>() {

                        @Override
                        public Row call(String v1) throws Exception {

                            return RowFactory.create(v1);

                        }

                    });

                    try {

                        StructType schema = new StructType(new StructField[]{
                            new StructField("trama", DataTypes.StringType, true, Metadata.empty())
                        });

                        DataFrame frame = sqlContext.createDataFrame(rddRow, schema);

                        if (frame.count() > 0) {
                            //Here is where the first messages throw the exception
                            frame.show();
                            frame.write().mode(SaveMode.Append).json("file:///C:/tmp/");

                        }

                    } catch (Exception ex) {

                        System.out.println(" INFO " + ex.getMessage());

                    }

                }

            });

            ssc.start();
            ssc.awaitTermination();

我无法更改spark的版本，因为此作业将在使用spark 1.6的旧cloudera集群中运行。我不知道我是做错了什么，还是只是个虫子。帮助

Answer 1

我解决了自己的问题，这个异常是由我如何创建SQLContext引发的，正确的方法是使用JavaStreamingContext创建SQLContext

//JavaStreamingContext jsc = ...
SQLContext sqlContext = new SQLContext(jsc.sparkContext());