代码之家 › 专栏 › 技术社区 › cscan ssice

将Spark结构化流输出写入Kafka主题

spark-structured-streaming apache-kafka apache-spark

cscan ssice · 技术社区 · 8 年前

SparkConf conf = new SparkConf()
        .setMaster("local[*]")
        .setAppName("test");

SparkSession spark = SparkSession
        .builder()
        .config(conf)
        .getOrCreate();

Dataset<Row> dataset = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", "start")
        .load();

StreamingQuery query = dataset
        .writeStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("checkpointLocation", "checkpoint")
        .option("topic", "end")
        .start();

query.awaitTermination(20000);

start end

1 回复 | 直到 7 年前

user6910411 7 年前

Dataset<Row> dataset = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", start.getTopicName())
        .option("startingOffsets", "earliest")
        .load();

推荐文章

Souhaib Guitouni · 将Spark结构化流媒体与融合模式注册表集成

8 年前

DeepNightTwo · 在Spark结构化流中使用自定义数据转换函数

8 年前

user1870400 · 如何在读取卡夫卡的消息流时处理Avro消息?

8 年前

djWann · 如何从AWS SQS读取流数据集?

8 年前

whisperstream · 在加载整个数据集之前,如何将自定义数据格式/映射应用于每个事件?

8 年前

Jochen Ullrich · 使用S3的Spark结构化流媒体无法连接

9 年前

Jacek Laskowski · 为什么id和runId在流查询的每次开始都会更改?

9 年前

M80 · 如何将from_json与模式作为字符串(即json编码的模式)一起使用?

9 年前

GOGO · “row+:savedState.toSeq”在StateStoreRestoreExec中做什么。doExecute?

9 年前

teddy · Spark 2.2.0流媒体无法找到数据源:kafka

9 年前