在Spark streaming中，是否可以将批量数据从kafka向上插入到Hive？

Question

我的计划是：

1. using spark streaming to load data from kafka every period like 1 minute.
2. convert the data loading every 1 min into DataFrame.
3. upsert the DataFrame into a Hive table (a table storing all history data)

目前，我成功地实现了步骤1-2。

我想知道是否有任何实用的方法来实现step3。详细说明：

1. load the latest history table with a certain partition in spark streaming.
2. use batch DataFrame to join the history table/DataFrame with a partition, and generate a new DataFrame.
3. save the new DataFrame to Hive, overwriting the history table of that partition.

下面是我的代码：

public final class SparkConsumer {
  private static final Pattern SPACE = Pattern.compile(" ");

  public static void main(String[] args) throws Exception {
    String brokers = "device1:9092,device2:9092,device3:9092";
    String groupId = "spark";
    String topics = "zhihu_comment";
    String destTable = "ods.zhihu_comment";

    // Create context with a certain seconds batch interval
    SparkConf sparkConf = new SparkConf().setAppName("TestKafkaStreaming");
    sparkConf.set("spark.streaming.backpressure.enabled", "true");
    sparkConf.set("spark.streaming.backpressure.initialRate", "10000");
    sparkConf.set("spark.streaming.kafka.maxRatePerPartition", "10000");
    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(60));

    Set<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));
    Map<String, Object> kafkaParams = new HashMap<>();
    kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

    kafkaParams.put("enable.auto.commit", false);
    kafkaParams.put("max.poll.records", "500");

    SparkSession spark = SparkSession.builder().appName(topics).getOrCreate();

    // Create direct kafka stream with brokers and topics
    JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(
            jssc,
        LocationStrategies.PreferConsistent(),
        ConsumerStrategies.Subscribe(topicsSet, kafkaParams));

    messages.foreachRDD(rdd -> {
        OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
        // some time later, after outputs have completed
        ((CanCommitOffsets) messages.inputDStream()).commitAsync(offsetRanges);
    });
    /*
     * Keep only the actual message in JSON format
     */
    Column[] colList =  { col("answer_id"), col("author"), col("content"), col("vote_count") };

    JavaDStream<String> recordStream = messages.flatMap(record -> Arrays.asList(record.value()).iterator());
    /*
     * Extract RDDs from stream
     */
    recordStream.foreachRDD(rdd -> {
        if (rdd.count() > 0) {
            Dataset<Row> df = spark.read().json(rdd.rdd());

            df.select(colList).show();
        }
    });

    // Get the lines, split them into words, count the words and print
    JavaDStream<String> lines = messages.map(ConsumerRecord::value);

    jssc.start();
    jssc.awaitTermination();
  }
}

我想知道这种方式是否可行？如果你能给我一些建议，我将不胜感激。

Answer 1

我强烈推荐Kafka Connect，而不是重新发明轮子。您所需要的就是HDFS Sink连接器，它可以将数据从Kafka主题复制到Hive：

Kafka Connect HDFS接收器允许您以各种格式将数据从Kafka主题导出到HDFS文件，并与配置单元集成，使数据可立即用于HiveQL查询

HDFS 2 Sink Connector

• For HDFS 2.x文件您可以使用HDFS 2 Sink Connector

• For HDFS 3.x文件使用HDFS 3 Sink Connector