spark读取kafka数据的Scala代码

答：可以使用Spark Streaming API将Kafka中的数据读取到Scala程序中，具体代码如下：val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "localhost:9092", // Kafka服务器地址 "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "use_a_separate_group_id_for_each_stream", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean) ) val topics = Array("topicA", "topicB") val stream = KafkaUtils.createDirectStream[String, String]( streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams) )

相关问题

sparkStreaming获取kafka数据（java版本）

以下是使用Java编写的Spark Streaming获取Kafka数据的示例代码：

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

public class KafkaSparkStreamingExample {

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

        // Kafka相关配置
        String brokers = "localhost:9092";
        String groupId = "test-group";
        String topics = "test-topic";

        // Spark相关配置
        SparkConf conf = new SparkConf().setAppName("KafkaSparkStreamingExample").setMaster("local[*]");
        JavaStreamingContext streamingContext = new JavaStreamingContext(conf, Durations.seconds(5));

        // Kafka参数
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", brokers);
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", groupId);
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        // 订阅Kafka主题
        Collection<String> topicsSet = Arrays.asList(topics.split(","));
        JavaInputDStream<Tuple2<String, String>> messages = KafkaUtils.createDirectStream(
                streamingContext,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, String>Subscribe(topicsSet, kafkaParams)
        );

        // 处理消息
        JavaPairDStream<String, String> pairs = messages.mapToPair(record -> new Tuple2<>(record.key(), record.value()));
        pairs.foreachRDD(rdd -> rdd.foreach(record -> System.out.println(record._2)));

        // 启动流处理
        streamingContext.start();
        streamingContext.awaitTermination();
    }
}

在这个示例代码中，我们使用了`KafkaUtils.createDirectStream()`方法订阅了一个Kafka主题，然后使用`mapToPair()`方法将消息转换为键值对形式的RDD，最后使用`foreachRDD()`方法遍历RDD并打印出每条记录的值。

spark读取oracle写入kafka,sparkStreaming读取kafka写入hive表

可以按照以下步骤实现：

1. 首先，需要在Spark中使用JDBC连接Oracle数据库，读取数据。可以使用以下代码：

val jdbcUrl = "jdbc:oracle:thin:@localhost:1521:ORCL"
val jdbcUsername = "username"
val jdbcPassword = "password"
val jdbcDriverClass = "oracle.jdbc.driver.OracleDriver"

val connectionProperties = new Properties()
connectionProperties.put("user", jdbcUsername)
connectionProperties.put("password", jdbcPassword)
connectionProperties.put("driver", jdbcDriverClass)

val oracleDF = spark.read.jdbc(jdbcUrl, "table_name", connectionProperties)

2. 接下来，将数据写入Kafka。可以使用以下代码：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger

val kafkaBrokers = "localhost:9092"
val kafkaTopic = "topic_name"

val kafkaDF = oracleDF.select(to_json(struct("*")).alias("value"))
  .selectExpr("CAST(NULL AS STRING) AS key", "value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaBrokers)
  .option("topic", kafkaTopic)
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()

3. 最后，使用Spark Streaming从Kafka读取数据，并将其写入Hive表。可以使用以下代码：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.sql.functions._

val spark = SparkSession.builder
  .appName("KafkaToHive")
  .enableHiveSupport()
  .getOrCreate()

val kafkaBrokers = "localhost:9092"
val kafkaTopic = "topic_name"

val schema = new StructType().add("column_name", StringType)

val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaBrokers)
  .option("subscribe", kafkaTopic)
  .option("startingOffsets", "earliest")
  .load()
  .select(from_json(col("value").cast("string"), schema).alias("data"))
  .select("data.*")

kafkaDF.writeStream
  .outputMode("append")
  .foreachBatch { (batchDF, _) =>
    batchDF.write.mode("append").insertInto("hive_table")
  }
  .start()
  .awaitTermination()

注意：在执行代码之前，需要先在Hive中创建相应的表，以便可以将数据写入其中。