连接 Flume 推特流以在 python 中激发时出现 UTF-8 编码错误

Question

我在将 Flume 代理收集的 Twitter 数据传递给 Spark Stream 时遇到问题。我可以独立下载推文，而只使用 Flume。但是我收到以下错误。我觉得这是关于 FlumeUtils.createStream() 中默认 UTF-8 编码的问题。我该如何改变它？我应该改变什么？

pyspark 终端错误：

org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 172, in main
    process()
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py", line 167, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/streaming/flume.py", line 107, in func
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/streaming/flume.py", line 36, in utf8_decoder
    return s.decode('utf-8')
  File "/usr/lib/python2.7/encodings/utf_8.py", line 16, in decode
    return codecs.utf_8_decode(input, errors, True)
UnicodeDecodeError: 'utf8' codec can't decode byte 0xe4 in position 17: invalid continuation byte

    at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:193)
    at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:234)
    at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
    at org.apache.spark.scheduler.Task.run(Task.scala:86)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
17/01/01 15:36:41 WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 1, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):

PySpark 代码：

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.streaming.flume import FlumeUtils

ss = SparkSession.builder \
    .master("local[2]") \
    .appName("Stream_Analysis")\
    .config("spark.sql.crossJoin.enabled", "true") \
    .getOrCreate()

sc = ss.sparkContext

strm = StreamingContext(sc, 5)

flume = FlumeUtils.createStream(strm,"localhost", 9999)
flume.pprint()
strm.start()
strm.awaitTermination()

Cmd 启动 pyspark

spark-submit --jars ~/project/spark-streaming-flume-assembly_2.11-2.0.2.jar ~/project/news_stream_flume/news_stream_analysis.py localhost 9999

Flume 会议：

# Name the components on this agent
FlumeAgent.sources = Twitter
FlumeAgent.sinks = spark
FlumeAgent.channels = MemChannel

# Twitter source
FlumeAgent.sources.Twitter.type = org.apache.flume.source.twitter.TwitterSource
FlumeAgent.sources.Twitter.consumerKey = x
FlumeAgent.sources.Twitter.consumerSecret =  y
FlumeAgent.sources.Twitter.accessToken = z
FlumeAgent.sources.Twitter.accessTokenSecret = xx
FlumeAgent.sources.Twitter.keywords = flume, spark

FlumeAgent.sinks.spark.type = avro
FlumeAgent.sinks.spark.channel = memoryChannel
FlumeAgent.sinks.spark.hostname = localhost
FlumeAgent.sinks.spark.port = 9999
FlumeAgent.sinks.spark.batch-size = 1

# Use a channel which buffers events in memory
FlumeAgent.channels.MemChannel.type = memory
FlumeAgent.channels.MemChannel.capacity = 10000
FlumeAgent.channels.MemChannel.transactionCapacity = 100

# Bind the source and sink to the channel
FlumeAgent.sources.Twitter.channels = MemChannel
FlumeAgent.sinks.spark.channel = MemChannel

Cmd 到运行 flume 代理：

flume-ng agent --name FlumeAgent --conf-file  /home/hduser/project/flume_config_2src_spark_avro  -f /usr/lib/flume-ng/conf/flume-conf.properties -Dflume.root.logger=DEBUG,console

Answer 1

FlumeUtils.createStream 接受一个 bodyDecoder 参数，这是一个用于字符串解码的函数。默认实现只检查 None 和 decodes 到 UTF-8:

def utf8_decoder(s):
    """ Decode the unicode as UTF-8 """
    if s is None:
        return None
    return s.decode('utf-8')

在Python 2.x中你应该可以用你自己的编码替换它，甚至完全跳过带有身份的解码（lambda x: x） .
Python 3.x 可能需要一些额外的步骤（使用 _.getBytes 的 JVM 端映射）来解决 String -> unicode 在 Pyrolite 中映射。

连接 Flume 推特流以在 python 中激发时出现 UTF-8 编码错误

UTF-8 encoding error while connecting Flume twitter stream to spark in python

apache-spark

pyspark

spark-streaming

flume-ng

flume-twitter