如何将 Spark sql 查询的所有输出保存到文本文件中

Question

我正在使用Spark流编写一个简单的消费者程序。我的代码将一些数据保存到文件中，但不是所有数据。谁能帮我解决这个问题。我不确定在哪里丢失了数据。我从 kafka 主题获取数据，然后我从 java Bean 类应用我的模式。

public class ConsumerFile {
public static void main(String[] args){
Logger.getLogger("org").setLevel(Level.OFF);
Logger.getLogger("akka").setLevel(Level.OFF);
String topic = args[0];
final String path=new String(args[2]);
String broker = args[1];
SparkConf sparkConf = new SparkConf().setAppName("SparkConsumer").setMaster("local[*]");;
    JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
    HashSet<String> topicsSet = new HashSet<String>(Arrays.asList(topic.split(",")));
    HashMap<String, String> kafkaParams = new HashMap<String, String>();
    kafkaParams.put("metadata.broker.list", broker);
    JavaPairInputDStream<String, String> kafkaStream = KafkaUtils.createDirectStream(
    ssc, String.class, String.class,StringDecoder.class,StringDecoder.class,kafkaPrams,
topicsSet
);
JavaDStream<String> words = kafkaStream.map(new Function<Tuple2<String, String>, String>() 
                                         {
                   public String call(Tuple2<String, String> message)
                                             {
                                                 return message._2();}});
words.foreachRDD(
          new Function2<JavaRDD<String>, Time, Void>() {
       public Void call(JavaRDD<String> rdd, Time time) {
   SQLContext sqlContext = JavaSQLContextSingleton.getInstance(rdd.context());
              // Convert RDD[String] to RDD[case class] to DataFrame
              JavaRDD<JavaRow> rowRDD = rdd.map(new Function<String, JavaRow>() {
                public JavaRow call(String line) throws Exception{
                String[] fields = line.split(",");
                  JavaRow record = new JavaRow(fields[0], fields[1],fields[2]  );
                  return record;
                }
              });
              DataFrame wordsDataFrame = sqlContext.createDataFrame(rowRDD, JavaRow.class);
              wordsDataFrame.registerTempTable("Data");
              DataFrame wDataFrame = sqlContext.sql(" select * from Data");  
              if(!wDataFrame.rdd().isEmpty()){
             wDataFrame.rdd().coalesce(1,true,null).saveAsTextFile(path); }
              return null;
            }} );
ssc.start();
 ssc.awaitTermination();}

}

Answer 1

这可能是因为您没有指定写入模式。而是使用它，

df.write.mode('append').text("/path/to/file")

PS：我不习惯用java做，我给出的是一个scala/python等价物

Answer 2

val sqlContext = new HiveContext(sc)
val df = sqlContext.sql("select * from tableName)
df.write.text("/path/to/file")

它将被写入为分区文本文件，因此您将结果间隔在一堆标记为 part-00000 的文件之间，但它会在那里。

Answer 3

我发现了它为什么这样做，以防其他人遇到同样的问题。 当你做foreachRDD时，它基本上在DStream的每个RDD上执行你的函数，你把它全部保存到同一个文件中。因此，它们会覆盖彼此的数据，第一个或最后一个写入者获胜。最简单的解决方法是将它们保存在具有唯一名称的文件中。所以我使用了saveAsTextFile（path + time（）.milliseconds（）.toString（））并解决了这个问题。但是，您可以两次使用相同的时间戳，因此我通过添加一个随机数使其更加独特。