ansible 一些基本使用方法的记录

使用 updateStateByKey 操作实现基于缓存的实时wordcount程序

updateStateByKey 介绍

updateStateByKey操作，可以为每个key维护一份state，并持续不断的更新该state。

1. 首先，要定义一个state，可以是任意的数据类型；
2. 其次，要定义state更新函数——指定一个函数如何使用之前的state和新值来更新state。

对于每个batch，Spark都会为每个之前已经存在的key去应用一次state更新函数，无论这个key在batch中是否有新的数据。如果state更新函数返回none，那么key对应的state就会被删除。当然，对于每个新出现的key，也会执行state更新函数。

注意，updateStateByKey操作，要求必须开启Checkpoint机制。

实例

from pyspark.streaming import StreamingContext
from pyspark import SparkContext

sc = SparkContext("local[2]", "streaming_socket")
ssc = StreamingContext(sc, 10)
# 必须开启 checkpoint 机制,这样才能把每个 key 对应的 state 存到内存和磁盘中, 以便于在内存数据丢失的时候, 可以从 checkpoint 中恢复
ssc.checkpoint("hdfs://127.0.0.1:9000/wordcount_checkpoint")

lines = ssc.socketTextStream("localhost", 9999)
words = lines\
    .flatMap(lambda l: l.split())\
    .map(lambda w: (w, 1))


# 第一个参数表示当前 batch 中这个 key 的值(可能有多个)
# 第二个参数表示 key 之前的 state
def word_count(values, state):
    if not state:
        state = 0
    for value in values:
        state += value
    return state


wordcount = words.updateStateByKey(word_count)
wordcount.pprint()

ssc.start()
ssc.awaitTermination()