Spark写入kafka(批数据和流式)

这篇具有很好参考价值的文章主要介绍了Spark写入kafka(批数据和流式)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Spark写入(批数据和流式处理)

Spark写入kafka批处理

写入kafka基础

Spark写入kafka(批数据和流式),Spark阶段,spark,kafka,linq

# spark写入数据到kafka
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()

# 创建df数据
df = ss.createDataFrame([[9, '王五', 21, '男'], [10, '大乔', 20, '女'], [11, '小乔', 22, '女']],
                        schema='id int,name string,age int,gender string')

df.show()
# todo 注意一:需要拼接一个value
# 在写入kafka时需要拼接一个value
df_kafka = df.select(F.concat_ws(',',df.id.cast('string'),df.name,df.age.cast('string'),df.gender).alias('value'))
df_kafka.show()

# 将df写入kafka
# todo 注意二:这个和读取kafka时的配置是一样,不过这里应该是没有读取起始量和读取结束量
options = {
    # 指定kafka的连接的broker服务节点信息
    'kafka.bootstrap.servers': 'node1:9092',
    # 指定写入主题
    'topic': 'user'
}
df_kafka.write.save(format='kafka', mode='append', **options)

kafka写入策略

Spark写入kafka(批数据和流式),Spark阶段,spark,kafka,linq文章来源地址https://www.toymoban.com/news/detail-824540.html

# kafka数据写入策略
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()


# 创建df数据
df = ss.createDataFrame([[200, '王五22222', 21, '男'], [201, '大乔22222', 20, '女'], [202, '小乔2222', 22, '女']],
                        schema='id int,name string,age int,gender string')

df.show()

# 在写入kakfa时需要拼接一个value
# # df_kafka = df.select(F.concat_ws(',',df.id.cast('string'),df.name,df.age.cast('string'),df.gender).alias('value'),F.lit(1).alias('partition'))
# # df_kafka.show()

# 指定分区 增加一个分区字段
options = {
    # 指定kafka的连接的broker服务节点信息
    'kafka.bootstrap.servers': 'node1:9092',
    # 指定写入主题
    'topic': 'user',
}
# df_kafka.write.save(format='kafka', mode='append', **options)



# 指定key  会key进行hash计算,相同key的数据会写入同一分区
# hash(key)%分区数  =
# df_kafka = df.select(F.concat_ws(',',df.id.cast('string'),df.name,df.age.cast('string'),df.gender).alias('value'),df.gender.alias('key'))
# df_kafka.show()

# 同时指定key和partition  按照分区写入
df_kafka = df.select(F.concat_ws(',',df.id.cast('string'),df.name,df.age.cast('string'),df.gender).alias('value'),df.gender.alias('key'),F.lit(2).alias('partition'))
df_kafka.show()

df_kafka.write.save(format='kafka', mode='append', **options)


写入kafka应答响应级别

# spark写入数据到kafka
# 指定ack应答级别
from pyspark.sql import SparkSession, functions as F

ss = SparkSession.builder.getOrCreate()

# 创建df数据
df = ss.createDataFrame([[9, '王五', 21, '男'], [10, '大乔', 20, '女'], [11, '小乔', 22, '女']],
                        schema='id int,name string,age int,gender string')

df.show()

# 在写入kakfa时需要拼接一个value
df_kafka = df.select(F.concat_ws(',', df.id.cast('string'), df.name, df.age.cast('string'), df.gender).alias('value'))
df_kafka.show()

# 将df写入kafka
options = {
    # 指定kafka的连接的broker服务节点信息
    'kafka.bootstrap.servers': 'node1:9092',
    # 指定写入主题
    'topic': 'user',
    # 指定级别
    'acks':'all'
}
df_kafka.write.save(format='kafka', mode='append', **options)

Sprak写入kafka流式处理

到了这里,关于Spark写入kafka(批数据和流式)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • ​理解 Spark 写入 API 的数据处理能力

    这张图解释了 Apache Spark DataFrame 写入 API 的流程。它始于对写入数据的 API 调用,支持的格式包括 CSV、JSON 或 Parquet。流程根据选择的保存模式(追加、覆盖、忽略或报错)而分岔。每种模式执行必要的检查和操作,例如分区和数据写入处理。流程以数据的最终写入或错误结束

    2024年02月03日
    浏览(47)
  • 在Spring Boot中使用Spark Streaming进行实时数据处理和流式计算

    引言: 在当今大数据时代,实时数据处理和流式计算变得越来越重要。Apache Spark作为一个强大的大数据处理框架,提供了Spark Streaming模块,使得实时数据处理变得更加简单和高效。本文将深入浅出地介绍如何在Spring Boot中使用Spark Streaming进行实时数据处理和流式计算,并提供

    2024年03月27日
    浏览(48)
  • 手把手入门MO | 如何使用使用 Spark 将批量数据写入 MatrixOne

    Apache Spark 是一个为高效处理大规模数据而设计的分布式计算引擎。它采用分布式并行计算的方式,将数据拆分、计算、合并的任务分散到多台计算机上,从而实现了高效的数据处理和分析。 大规模数据处理与分析 Spark 能够处理海量数据,通过并行计算任务提高了处理效率。

    2024年02月01日
    浏览(188)
  • 【Spark大数据习题】习题_Spark SQL&&&Kafka&& HBase&&Hive

    PDF资源路径-Spark1 PDF资源路径-Spark2 一、填空题 1、Scala语言的特性包含面向对象编程、函数式编程的、静态类型的、可扩展的、可以交互操作的。 2、在Scala数据类型层级结构的底部有两个数据类型,分别是 Nothing和Null。 3、在Scala中,声明变量的有var声明变量和val声明常

    2024年02月06日
    浏览(44)
  • 问题:Spark SQL 读不到 Flink 写入 Hudi 表的新数据,打开新 Session 才可见

    博主历时三年精心创作的《大数据平台架构与原型实现:数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行,点击《重磅推荐:建大数据平台太难了!给我发个工程原型吧!》了解图书详情,京东购书链接:https://item.jd.com/12677623.html,扫描左侧

    2024年02月22日
    浏览(55)
  • Spark Streaming + Kafka构建实时数据流

    1. 使用Apache Kafka构建实时数据流 参考文档链接:https://cloud.tencent.com/developer/article/1814030 2. 数据见UserBehavior.csv 数据解释:本次实战用到的数据集是CSV文件,里面是一百零四万条淘宝用户行为数据,该数据来源是阿里云天池公开数据集 根据这一csv文档运用Kafka模拟实时数据流,

    2024年02月12日
    浏览(44)
  • 项目实战——参数配置化Spark将Hive表的数据写入需要用户名密码认证的ElasticSearch(Java版本)

    项目实战——将Hive表的数据直接导入ElasticSearch    此篇文章不用写代码,简单粗暴,但是相对没有那么灵活;底层采用MapReduce计算框架,导入速度相对较慢! 项目实战——Spark将Hive表的数据写入ElasticSearch(Java版本)    此篇文章需要Java代码,实现功能和篇幅类似,直接

    2023年04月08日
    浏览(62)
  • Spark解析JSON文件,写入hdfs

    一、用Sparkcontext读入文件,map逐行用Gson解析,输出转成一个caseclass类,填充各字段,输出。 解析JSON这里没有什么问题。 RDD覆盖写的时候碰到了一些问题 : 1.直接saveAsTextFile没有覆盖true参数; 2.转dataframe时,还得一个一个字段显化才能转成dataframe; 3.write时,一开始打算写

    2024年01月23日
    浏览(41)
  • spark读取、写入Clickhouse以及遇到的问题

    最近需要处理Clickhouse里面的数据,经过上网查找总结一下spark读写Clickhouse的工具类已经遇到的问题点。具体Clickhouse的讲解本篇不做讲解,后面专门讲解这个。 话不多说直接看代码 1.引入依赖: 0.2.4  这个版本用的比较多一点 2.spark对象创建  3.spark读取clickhouse数据:

    2024年02月03日
    浏览(45)
  • 【大数据技术】Spark+Flume+Kafka实现商品实时交易数据统计分析实战(附源码)

    需要源码请点赞关注收藏后评论区留言私信~~~ 1)Kafka 是一个非常通用的系统,你可以有许多生产者和消费者共享多个主题Topics。相比之下,Flume是一个专用工具被设计为旨在往HDFS,HBase等发送数据。它对HDFS有特殊的优化,并且集成了Hadoop的安全特性。如果数据被多个系统消

    2024年02月03日
    浏览(53)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包