Как объединить два DStreams (pyspark)?

Question

У меня есть поток кафки, входящий в тему ввода. Это код, который я написал для принятия потока кафки.

conf = SparkConf().setAppName(appname) 
sc = SparkContext(conf=conf) 
ssc = StreamingContext(sc) 
kvs = KafkaUtils.createDirectStream(ssc, topics,\ 
      {"metadata.broker.list": brokers}) 

Затем я создаю два DStreams ключей и значений исходного потока.

keys = kvs.map(lambda x: x[0].split(" ")) 
values = kvs.map(lambda x: x[1].split(" ")) 

Затем я выполняю вычисления в значениях DStream. Для примера,

val = values.flatMap(lambda x: x*2) 

Теперь, мне нужно объединить ключи и VAL DStream и возвращает результат в виде Кафки потока.

Как совместить val с corressponding ключом?

Answer 1

Вы можете просто использовать оператор join на 2 DStreams, чтобы объединить их. Когда вы делаете карту, вы по существу создаете другой поток. Таким образом, объединение поможет вам объединить их вместе.

Например:

Joined_Stream = keys.join(values).(any operation like map, flatmap...)