PySpark Structured Streaming kafka示例

爱被打了一巴掌 2022-04-22 15:06 317阅读 0赞

## PySpark Structured Streaming kafka示例 ##

直接上代码：

#!/usr/bin/env python3      
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import from_json
    from pyspark.sql.types import StructType, StringType
    
    if __name__ == '__main__':
        spark = SparkSession \
            .builder \
            .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0") \
            .appName("pyspark_structured_streaming_kafka") \
            .getOrCreate()
    
        df = spark.readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", "localhost:9092") \
            .option("subscribe", "test") \
            .load()
    
        words = df.selectExpr("CAST(value AS STRING)")
    
        schema = StructType() \
            .add("name", StringType()) \
            .add("age", StringType()) \
            .add("sex", StringType())
        
        # 通过from_json，定义schema来解析json
        res = words.select(from_json("value", schema).alias("data")).select("data.*")
    
        query = res.writeStream \
            .format("console") \
            .outputMode("append") \
            .start()
    
        query.awaitTermination()

程序启动后，在kafka的product命令行，输入`{"name":"caocao","age":"32","sex":"male"}`，在控制台输入：

![image][]

[image]: /images/20220327/35df5f6d9b1e4153945d7acc70d58248.png

发表评论取消回复

表情：

评论列表（有 0 条评论，317人围观）

还没有评论，来说两句吧...

相关阅读

相关七、Structured Streaming Kafka

1、pom.xml <properties> <project.build.sourceEncoding>UTF-8</project.bui...

朱雀/ 2024年04月18日 16:01/ 0 赞/ 4 阅读

相关 Structured Streaming详解

目录一、Structured Streaming概述（1）Structured Streaming背景（2）S

àì夳堔傛蜴生んèń/ 2022年12月29日 11:27/ 0 赞/ 95 阅读

相关 Structured Streaming架构原理详解！

目录一、Structured Streaming概述二、Structured Streaming与Socket集成三、Str

妖狐艹你老母/ 2022年12月11日 12:13/ 0 赞/ 184 阅读

相关 Structured Streaming

Spark2.0新增了Structured Streaming，它是基于SparkSQL构建的可扩展和容错的流式数据处理引擎，使得实时流式数据计算可以和离线计算采用相同的处理方

阳光穿透心脏的1/2处/ 2022年06月02日 23:26/ 0 赞/ 103 阅读

相关 PySpark Structured Streaming kafka示例

PySpark Structured Streaming kafka示例直接上代码： !/usr/bin/env python3 --

爱被打了一巴掌/ 2022年04月22日 15:06/ 0 赞/ 318 阅读

相关 Structured Streaming入门实例

Structured Streaming入门实例 Structured Streaming是Spark2.4版本推出的新的实时流处理计算。相比Spark Streamin

旧城等待，/ 2022年03月28日 12:48/ 0 赞/ 145 阅读

相关 Kafka Stream简单示例（一）

最近想统计一些消息数据，原计划接收kakfa消息后自行统计然后存入数据库（统计相对比较简单，所以没有考虑使用Apache Storm），突然想起来Kafka已经提供Kakfa

拼搏现实的明天。/ 2022年03月15日 15:11/ 0 赞/ 210 阅读

相关 Spark Structured Streaming

With the ubiquity of real-time data, organizations need streaming systems that are scala

雨点打透心脏的1/2处/ 2022年01月20日 11:59/ 0 赞/ 109 阅读

相关 Spark Structured Streaming + Kafka使用笔记

这篇博客将会记录Structured Streaming + Kafka的一些基本使用(Java 版) 1. 概述 Structured Streaming （结构化流

电玩女神/ 2021年12月21日 02:57/ 0 赞/ 347 阅读

相关 Spark Structured Streaming + Kafka使用笔记

这篇博客将会记录Structured Streaming + Kafka的一些基本使用(Java 版) 1. 概述 Structured Streaming （结构化流

怼烎@/ 2021年12月21日 02:41/ 0 赞/ 277 阅读