spark streaming 将数据持久化的数据库
spark streaming 将数据持久化的数据库
查看资料与官网,都没有找到streaming入mysql的python程序,无奈自己写了一个(写了半天,才测试成功,真实太菜了),希望能够帮助到各位。。。
启动pyspark
[[email protected] bin]# ./pyspark --master local[2]
程序如下:
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
import pymysql
##使用sparkstreaming将数据存储到mysql
ssc = StreamingContext(sc,5)
#使用socket生成数据
lines = ssc.socketTextStream(“localhost”,6789)
#词频统计
counts = lines.flatMap(lambda line:line.split(" "))
.map(lambda word:(word,1))
.reduceByKey(lambda a,b:a+b)
#创建mysql链接
def createConnection():
conn = pymysql.Connect(host=“localhost”,user=“root”,password=“root”,database=“sparkstreaming”,port=3306)
return conn
#将数据推送到mysql
def sendPartition(iter):
connection = createConnection()
for record in iter:
sql = “insert into wordcount(word, wordcount) values(’%s’” % record[0] +’,’+"%s)" % record[1]
connection.cursor().execute(sql)
connection.commit()
connection.close()
counts.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))
counts.pprint()
ssc.start()
ssc.awaitTermination()
启动nc,发送数据
查看控制台,数据输出
查看mysql,数据入库