网上用python写spark+kafka的资料好少啊 自己记录一点踩到的坑~
spark+kafka介绍的官方网址：http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html
python的pyspark库函数文档：http://spark.apache.org/docs/latest/api/python/pyspark.streaming.html?highlight=kafkautils.createdirectstream#pyspark.streaming.kafka.KafkaUtils.createDirectStream
上面两个是最重要的资料，大多数问题可以通过仔细研读上面两个文档得到答案
官网上说了，spark和kafka连用有两种方式：接收器形式 以及 直连形式
一、 接收器形式
优点：支持kafka的group.id设置，支持用kafka api查询offset，如果数据断掉后，可以通过group.id轻松找到上一次失败的位置
缺点：
1.失败处理复杂。由于kafka队列信息由kafka自己记录，当spark消费了数据但是处理中出错时会导致数据丢失。为了避免数据丢失就必须开启Write Ahead Logs，把spark接收到的数据都存储到分布式文件系统中，比如HDFS，然后失败时从存储的记录中找到失败的消息。这导致同一批数据被kafka和spark存储了2次。造成数据冗余。
2.如果有多个地方都想获取同一个kafka队列的数据，必须建立多个流，无法用一个流并行处理。
该方法是比较老的一种方式，并不太被推荐。
二、直连形式
优点：
1. 不需两次存储数据，直连形式时，spark自己管理偏移信息，不再使用kafka的offset信息。所以spark可以自行处理失败情况，不要再次存储数据。spark保证数据传输时Exactly-once。
2.只需建立一个流就可以并行的在多个地方使用流中的数据
缺点：
不支持kafka的group，不支持通过kafka api查询offset信息！！！！
在连接后spark会根据fromOffsets参数设置起始offset，默认是从最新的数据开始的。也就是说，必须自己记录spark消耗的offset位置。否则在两次脚本启动中间的数据都会丢失。
我选用的是直连形式，我处理offset的方法是将spark消费的offset信息实时记录到文件中。在启动脚本时通过记录的文件来找到起始位置。
#!/usr/bin/python
# coding=utf-8
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
import time
import os
import json
broker_list = "xxxx"
topic_name = "xxxx"
timer = 5
offsetRanges = [] def store_offset_ranges(rdd):
global offsetRanges
offsetRanges = rdd.offsetRanges()
return rdd def save_offset_ranges(rdd):
root_path = os.path.dirname(os.path.realpath(__file__))
record_path = os.path.join(root_path, "offset.txt")
data = dict()
f = open(record_path, "w")
for o in offsetRanges:
data = {"topic": o.topic, "partition": o.partition, "fromOffset": o.fromOffset, "untilOffset": o.untilOffset}
f.write(json.dumps(data))
f.close() def deal_data(rdd):
data = rdd.collect()
for d in data:
# do something
pass def save_by_spark_streaming():
root_path = os.path.dirname(os.path.realpath(__file__))
record_path = os.path.join(root_path, "offset.txt")
from_offsets = {}
# 获取已有的offset，没有记录文件时则用默认值即最大值
if os.path.exists(record_path):
f = open(record_path, "r")
offset_data = json.loads(f.read())
f.close()
if offset_data["topic"] != topic_name:
raise Exception("the topic name in offset.txt is incorrect") topic_partion = TopicAndPartition(offset_data["topic"], offset_data["partition"])
from_offsets = {topic_partion: long(offset_data["untilOffset"])} # 注意设置起始offset时的方法
print "start from offsets: %s" % from_offsets sc = SparkContext(appName="Realtime-Analytics-Engine")
ssc = StreamingContext(sc, int(timer)) kvs = KafkaUtils.createDirectStream(ssc=ssc, topics=[topic_name], fromOffsets=from_offsets,
kafkaParams={"metadata.broker.list": broker_list})
kvs.foreachRDD(lambda rec: deal_data(rec))
kvs.transform(store_offset_ranges).foreachRDD(save_offset_ranges) ssc.start()
ssc.awaitTermination()
ssc.stop() if __name__ == '__main__':
save_by_spark_streaming()
运行：
正常情况下，只要输入下面的语句就可以运行了
spark-submit --packages org.apache.spark:spark-streaming-kafka--8_2.:2.2. spark_kafka.py
然而，我的总是报错，找不到依赖包，说各种库不认识。所以我只好用--jars来手动指定包的位置了..................
spark-submit --packages org.apache.spark:spark-streaming-kafka--8_2.:2.2. --jars /root/.ivy2/jars/org.apache.kafka_kafka_2.-0.8.2.1.jar,/root/.ivy2/jars/com.yammer.metrics_metrics-core-2.2..jar spark_kafka.py
吐槽：
我就踩在直连形式不支持offset的坑上了..... 开始官方文档没仔细看，就瞄了一眼说是直连形式好，就豪不犹豫的用了。结果我的脚本不稳定，各种断，然后中间数据就各种丢啊.......
还有官网上居然完全没有对fromOffsets这个参数的说明，我找了好久好久才弄清楚这个参数怎么拼出来啊.................