from pyspark import SparkConf, SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import json
import time
sc = SparkContext(master = "local[2]",appName="StreamingWordCount")
ssc = StreamingContext(sc, 5)
# HBASE表，需要提前在HBASE中建好
table = 'flume'
broker = "192.168.159.148:9094"
# kafka的topic
topic = "hbasespark"
# HBASE的zookeeper
hbaseZK = "192.168.159.148"
keyConv = "org.apache.spark.examples.pythonconverters.StringToImmutableBytesWritableConverter"
valueConv = "org.apache.spark.examples.pythonconverters.StringListToPutConverter"
hbaseConf = {"hbase.zookeeper.quorum": hbaseZK, "hbase.mapred.outputtable": table,
"mapreduce.job.output.value.class": "org.apache.hadoop.io.Writable"}
#列印日志
def log(str):
t = time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime())
print("[%s]%s" % (t, str))
#處理RDD元素，此RDD元素需為字典型別
def fmt_data(msg_dict):
if msg_dict is not None:
t = time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime())
rowkey =t
lst = []
for d, x in msg_dict.items():
col_name = d
col_value = str(x)
col_family = 'c1'
msg_tuple = (rowkey, [rowkey, col_family, col_name, col_value])
print("rowkey:" + rowkey + "\ndata " + str(msg_tuple) + " append success")
if msg_tuple is not None:
lst.append(msg_tuple)
return lst
#處理RDD并向HBASE中寫入
def connectAndWrite(data):
if not data.isEmpty():
# 接收到的RDD中的元素轉為字典，收到的格式為(None,[json串])，所以map第二個元素反序列化成為字典型別
msg_list = data.map(lambda x: json.loads(x[1]))
# 列印RDD觀察，類似一個以字典型別資料為元素的串列
log(msg_list.collect())
try:
# 處理RDD中元素為寫入HBASE需要的格式，形成元組格式
msg_row = msg_list.map(lambda x: fmt_data(x))
# print(msg_row.flatMap(lambda x: x).map(lambda x: x).collect())
# 將RDD中所有元素中的元組扁平化，再map后往HBASE存盤
msg_row.flatMap(lambda x: x).map(lambda x: x).saveAsNewAPIHadoopDataset(conf=hbaseConf, keyConverter=keyConv,valueConverter=valueConv)
print("插入資料成功")
except Exception as ex:
print(str(ex) + " 插入資料失敗")
kafkaStreams = KafkaUtils.createDirectStream(ssc, [topic], kafkaParams={"metadata.broker.list": broker})
#kafkaStreams.map(lambda x:x[1]).pprint()
kafkaStreams.foreachRDD(connectAndWrite)
log('start consumer')
ssc.start()
ssc.awaitTermination()
運行之后會報
18/06/06 10:07:20 INFO python.Converter: Loaded converter: org.apache.spark.examples.pythonconverters.StringListToPutConverter
An error occurred while calling z:org.apache.spark.api.python.PythonRDD.saveAsHadoopDataset.
: java.lang.NullPointerException
這個錯誤的根源在哪里啊 求指教
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/50556.html
標籤：Spark
上一篇：【華為云社區 18年5月刊】本期推薦：華為云資料庫服務資料匯總；機器學習服務案例實踐；華為云精品訪談欄目“云視界”重磅開播！
下一篇：實體秘鑰檔案丟了 怎么辦