Apache Kafka/Spark/Streaming 方法

Question

这个问题可能是重复的。

我想从 Spark 收听一个 Kafka 主题并将内容传递给 Ignite Cache。 我喜欢实现Performance Tuning of an Apache Kafka/Spark Streaming System 中描述的相同目标。
使用KafkaUtils.createDirectStream() 在 Spark 中读取 Kafka 主题，使用IgniteRDD 将数据推送到 Ignite 缓存中。 但系统抛出如下错误：
org.apache.spark.SparkException: This RDD lacks a SparkContext. It could happen in the following cases: (1) RDD transformations and actions are NOT invoked by the driver, but inside of other transformations; for example, rdd1.map(x => rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063. (2) When a Spark Streaming job recovers from checkpoint, this exception will be hit if a reference to an RDD not defined by the streaming job is used in DStream operations. For more information, See SPARK-13758.
代码如下：

public static void main(String[] args) throws Exception{

        SparkConf conf = new SparkConf()
     .setAppName("kafka-sandbox")
                .setMaster("local[*]");
        conf.set("spark.driver.allowMultipleContexts", "true");
        JavaSparkContext sc = new JavaSparkContext(conf);

        //Context for Kafka
        JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));
        // Creates Ignite context with specific configuration and runs Ignite in the embedded mode.
        IgniteContext igniteContext = new IgniteContext(
            sc.sc(),"/home/ec2-user/apache-ignite-fabric-2.6.0-bin/config/default-config.xml", false);


        // Adjust the logger to exclude the logs of no interest.
        Logger.getRootLogger().setLevel(Level.ERROR);
        Logger.getLogger("org.apache.ignite").setLevel(Level.INFO);



        // Define data to be stored in the Ignite RDD (cache).
        List<Integer> data = new ArrayList<>(20);

        for (int i = 0; i<20; i++) {
            data.add(i);
        }



        Set<String> topics = Collections.singleton("Hello-Kafka");
        Map<String, String> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", "10.0.102.251:9092");

        JavaPairInputDStream<String, String> directKafkaStream = KafkaUtils.createDirectStream(ssc,
                String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);



        directKafkaStream.foreachRDD(rdd -> {






        // Create a Java Ignite RDD of Type (Int,Int) Integer Pair.
        IgniteRDD sharedRDD = igniteContext.fromCache("hello-spark");
        // Preparing a Java RDD
        JavaRDD<String> javaRDD =  sc.parallelize(Collections.singletonList("Hello-world"));
                        System.out.println("--- New RDD with " + rdd.partitions().size() + " partitions and " + rdd.count() + " records");
                        rdd.foreach(record -> {
                            //Displaying Kafka topic
                                System.out.println("Got the record : " + record._2);
                                //Pushing valeus to Ignite
                               sharedRDD.savePairs(javaRDD.<Integer, Integer>mapToPair(new PairFunction<Integer, Integer, Integer>() {
           @Override public Tuple2<Integer, Integer> call(Integer val) throws Exception {
                return new Tuple2<Integer, Integer>(val, val);
            }
        })
);
        ssc.start();
        ssc.awaitTermination();


                }
}

我无法找出代码中缺少的内容。 方法是正确的还是我应该使用另一种方法。 请指导我。

Answer 1

您的示例可能会简化为以下代码：

JavaRDD<Integer> rdd = sparkCtx.parallelize(Arrays.asList(1, 2, 3));
JavaRDD<Integer> javaRDD = sparkCtx.parallelize(Arrays.asList(4, 5, 6));
JavaIgniteRDD<Integer, Integer> sharedRDD = igniteCtx.fromCache("hello-spark");

rdd.foreach(record ->
    sharedRDD.savePairs(
        javaRDD.mapToPair((PairFunction<Integer, Integer, Integer>)val ->
            new Tuple2<>(val, val))
    ));

我从等式中删除了 Kafka 以简化示例。

首先，这很奇怪，您遍历rdd 的元素并将javaRDD 的值放入sharedRDD，同时忽略rdd 记录。其中rdd 和javaRDD 是不同的东西。我不明白，你为什么这样做。

您收到异常是因为您在 foreach 内运行 mapToPair 操作。两者都是RDD操作，不能嵌套。您应该将savePairs 部分移出foreach，或者以某种方式将rdd 和javaRDD 结合起来，这样就不需要运行嵌套的RDD 操作。这取决于您真正想要实现的目标。