Spark将Kafka输入流另存为Json文件

当您映射 JavaInputDStream 到a stream 您可以执行以下操作:

stream.foreachRDD(rdd -> {
            OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();

            rdd.mapToPair(new PairFunction<ConsumerRecord<String, String>, String, String>() {
                @Override
                public Tuple2<String, String> call(ConsumerRecord<String, String> record) {
                    return new Tuple2<>(record.key(), record.value());
                }
            }).foreachPartition(partition -> {

                OffsetRange o = offsetRanges[TaskContext.get().partitionId()];
                System.out.println(o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset());

                if (partition.hasNext()) {

                    PrintWriter out = new PrintWriter("filename.txt");;
                    out.println(text);

                    try {
                        while (partition.hasNext()) {

                            Tuple2<String, String> message = partition.next();
                            out.println(message);
                        }

                    } catch (Exception e) {
                        e.printStackTrace(
                }

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

不要忘记,如果您的Kafka主题中有多个分区,那么您将按照上述方法为每个分区编写一个文件。