sparkstreaming 判断不同的topic写入不同的hive表

在Spark Streaming中，可以通过一些条件判断来将不同的topic写入不同的Hive表。

首先，我们可以使用DStream的foreachRDD方法来处理每个批次的输入数据。在foreachRDD方法中，可以根据条件判断来选择对应的Hive表进行写入操作。

假设我们有两个topic：topic1和topic2。我们可以在foreachRDD方法中使用if-else语句来判断当前批次的输入数据属于哪个topic，然后分别将数据写入对应的Hive表。

具体操作如下：

1. 创建一个StreamingContext并设定批处理间隔。

val conf = new SparkConf().setAppName("SparkStreamingHive")
val ssc = new StreamingContext(conf, Seconds(5))

2. 创建一个DStream并从Kafka中读取数据。

val topics = Set("topic1", "topic2")
val kafkaParams = Map[String, String]("metadata.broker.list" -> "localhost:9092")
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

3. 在foreachRDD方法中，根据条件判断来选择对应的Hive表进行写入操作。

messages.foreachRDD { rdd =>
    if (rdd.isEmpty()) {
        println("No data to process")
    } else {
        if (rdd.topic() == "topic1") {
            // 将数据写入topic1对应的Hive表
            rdd.saveAsHiveTable("table1")
            println("Data from topic1 saved to Hive table1")
        } else if (rdd.topic() == "topic2") {
            // 将数据写入topic2对应的Hive表
            rdd.saveAsHiveTable("table2")
            println("Data from topic2 saved to Hive table2")
        }
    }
}

以上代码中，我们使用rdd.topic()来获取当前RDD的topic。然后我们可以使用if-else语句根据topic的值来选择对应的Hive表进行写入操作。

通过这样的方式，我们可以根据不同的topic将数据写入不同的Hive表。注意，以上代码仅为示例，实际操作中需要根据实际情况进行调整和优化。