Spark写入kafka（批数据和流式）

Spark写入（批数据和流式处理）

Spark写入kafka批处理

写入kafka基础

在这里插入图片描述

# 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写入策略

在这里插入图片描述

# 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)