依赖包

  1. <dependency>
  2.     <groupId>org.apache.spark</groupId>
  3.     <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
  4.     <version>2.3.0</version>
  5. </dependency>

代码

  1. import redis
  2. from redis.client import StrictRedis
  3. from pyspark.sql import SparkSession
  4. from pyspark.streaming import StreamingContext
  5. from pyspark.streaming.kafka import KafkaUtils
  6. from pyspark.streaming.kafka import OffsetRange, TopicAndPartition
  7. from pyspark import RDD
  8. import json
  9. import time
  11. redis_config = {
  12.     "host": "localhost",
  13.     "port": 6379,
  14.     "db": 3,
  15.     "password": "123456",
  16.     "redis_offset_key": "test_offset4",
  17.     "redis_stop_key": "stop_test"
  18. }
  20. kafka_config = {
  21.     "topic": ['my_topic'],
  22.     "kafka_params": {
  23.         "bootstrap.servers": "localhost:9092",
  24.         "group.id": "wsy3_group_id",
  25.         "auto.offset.reset": "smallest"
  26.     }
  27. }
  30. def get_redis_client(redis_config: dict) -> StrictRedis:
  31.     pool = redis.ConnectionPool(host=redis_config["host"], port=redis_config["port"], db=redis_config["db"],
  32.                                 password=redis_config["password"])
  33.     redis_client = redis.StrictRedis(connection_pool=pool)
  34.     return redis_client
  37. def get_redis_fromOffsets(redis_client: StrictRedis) -> dict:
  38.     redis_fromOffsets = None
  39.     redis_offset_key = redis_config['redis_offset_key']
  40.     existing_offset = redis_client.get(redis_offset_key)
  41.     if existing_offset:
  42.         print("-----------------从redis启动-----------------------")
  43.         # 一个redis中存储的数据样例
  44.         # [{"topic": "topic", "partition": 0, "fromOffset": 7953207, "untilOffset": 7953226},
  45.         # {"topic": "topic", "partition": 1, "fromOffset": 7951255, "untilOffset": 7951274}]
  46.         topic_partition_offset_list = json.loads(existing_offset.decode())
  47.         topicAndPartition_offset_dict = {}
  48.         for x in topic_partition_offset_list:
  49.             print("***************redis offset------------------")
  50.             print(x)
  51.             print("***************redis offset------------------")
  52.             topicAndPartition = TopicAndPartition(x['topic'], x['partition'])
  53.             offset = x['fromOffset']
  54.             topicAndPartition_offset_dict.update({
  55.                 topicAndPartition: offset
  56.             })
  57.         redis_fromOffsets = topicAndPartition_offset_dict
  58.     else:
  59.         print("--------------直接启动----------------------------")
  60.     return redis_fromOffsets
  63. def gracefull_stop(ssc: StreamingContext, redis_client: StrictRedis, redis_stop_key: str) -> None:
  64.     is_stop = False
  65.     while not is_stop:
  66.         print("------------------等待stop------------------------")
  67.         is_stop = ssc.awaitTerminationOrTimeout(30)
  68.         if not is_stop and redis_client.exists(redis_stop_key) and "stop" == redis_client.get(redis_stop_key).decode():
  69.             print("-----------10秒后开始stop-------------------")
  70.             time.sleep(10)
  71.             ssc.stop(True, True)
  74. def main():
  75.     spark = SparkSession.builder \
  76.         .master('local') \
  77.         .appName('write_offset_to_redis') \
  78.         .getOrCreate()
  79.     spark.sparkContext.setLogLevel('INFO')
  80.     ssc = StreamingContext(spark.sparkContext, 5)
  81.     redis_client = get_redis_client(redis_config)
  83.     redis_fromOffsets = get_redis_fromOffsets(redis_client)
  84.     directKafkaStream = KafkaUtils.createDirectStream(ssc, kafka_config['topic'], kafka_config['kafka_params'],
  85.                                                       fromOffsets=redis_fromOffsets)
  87.     def storeOffsetRanges(rdd: RDD) -> RDD:
  88.         offsetRanges = rdd.offsetRanges()
  89.         offset_redis = []
  90.         for o in offsetRanges:
  91.             offset = {
  92.                 "topic": o.topic,
  93.                 "partition": o.partition,
  94.                 "fromOffset": o.fromOffset,
  95.                 "untilOffset": o.untilOffset
  96.             }
  97.             offset_redis.append(offset)
  98.             print("%s %s %s %s" % (o.topic, o.partition, o.fromOffset, o.untilOffset))
  99.         redis_offset_value = json.dumps(offset_redis)
  100.         redis_offset_key = redis_config['redis_offset_key']
  101.         redis_client.set(redis_offset_key, redis_offset_value)
  102.         return rdd
  104.     def printOffsetRanges(rdd: RDD) -> None:
  105.         pass
  107.     directKafkaStream \
  108.         .transform(storeOffsetRanges) \
  109.         .foreachRDD(printOffsetRanges)
  111.     ssc.start()
  112.     gracefull_stop(ssc, redis_client, redis_config['redis_stop_key'])
  113.     ssc.awaitTermination()
  116. if __name__ == '__main__':
  117.     main()

启动shell脚本

  1. spark2-submit \
  2. --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.3.0 \
  3. --master local \
  4. /my/dir/h.py

建议

这个版本的太老,不建议使用
强烈推荐使用Structured Streaming,工程代码如下
Spark redis 管理 kafka offset 和优雅停止