Spark笔记15-Spark数据源及操作

# 创建文件存放的目录
cd /usr/loca/spark/mycode
mkdir streaming
cd streaming
mkdir logfile
cd logfile   # 对这个子目录进行数据监控

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
ssc = StreamingContext(sc, 10)   # 每10秒监听；交互式环境下自带sc实例对象
lines = ssc.textFileStream(".../logfile")  # 创建文件流，监控目录的全称地址
words = lines.flatMap(lambda line:line.split(' '))  # 通过flatMap操作将数据进行lambda操作，再进行拍平
wordCounts = words.map(lambda x:(x,1)).reduceByKey(lambda a,b: a+b)
wordCounts.pprint()  # 在交互式环境下查看
ssc.start()   # 启动流计算
ssc.awaitTermination()  # 等待流计算结束

# 创建文件存放的目录
cd /usr/loca/spark/mycode
mkdir streaming
cd streaming
mkdir socket
cd socket

# vim NetworkWordCount.py：扮演的是客户端角色

from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == "__main__":
  if len(sys.argv) != 3:   # 第一个参数默认是self
    print("Usage: NetworkWordCount.py<hostname><port>", file=sys.stderr)
    exit(-1)   # 参数长度不够，自动退出
  sc = SparkContext(appName="pythonStreamingNetworkWordCount")
  ssc = StreamingContext(sc, 1)   # 流计算的指挥官
  lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))  # 定义套接字类型的输入源
  counts = lines.flatMap(lambda line:line.split(" ").map(lambda word:(word,1)).reduceByKey(lambda a,b: a+b)

  counts.pprint()
  ssc.start()
  ssc.awaitTermination()


# 服务端的角色
# 在linux中：nc -lk 9999
cd /usr/local/spark/mycode/streaming/socket
/usr/local/spark/bin/spark-submit NetworkWordCount.py localhost 9999

# 使用socket编程实现自定义数据源
# DataSourceSocket.py

import socket
server = socket.socket()  #  生成对象
server.bind("localhose", 9999)   # 设置监听的机器和端口号
server.listen(1)
while 1:
  conn,addr = server.accept()  # 使用两个值进行接受
  print("connect success! connection is from %s" %addr[0])
  print("sending data....")
  conn.send("I love hadoop  I love spark hadoop is good spark is fast".encode())  # 打印正在传输的数据
  conn.close()
  print("connection is broken.")

cd /usr/local/spark/mycode/streaming/socket
/usr/local/spark/bin/spark-submit DataSourceSocket.py

# 启动客户端
cd /usr/local/spark/mycode/streaming/socket
/usr/local/spark/bin/spark-submit NetworkWordCount.py localhost 9999

cd /usr/local/spark/mycode/streaming/rddqueue
/usr/local/spark/bin/spark-submit RDDQueueStream.py

# RDDQueueStream.py

import time
from pyspark import SparkContext
from pyspark.Streaming import StreamingContext

if __name__ == "__main__":
  sc = sparkContext(appName="pythonStreamingQueueStream")
  ssc = StreamingContext(sc, 2)   # 数据流指挥官的生成
  rddQueue = []
  for i in range(5):
    rddQueue += [ssc.sparkContext.parallelize[j for j in range(1,1001)], 10]   #. 创建RDD队列流
    time.sleep(1)

  inputStream = ssc.queueStream(rddQueue)
  mappedStream = inputStream.map(lambda x:(x %10, 1))
  reduceStream = mappedStream.reduceByKey(lambda a,b: a + b)
  reduceStream.pprint()
  ssc.start()
  ssc.stop(stopSparkContext=True, stopGraceFully=True)

# 将下载解压后的jar包拷贝到spark的jars目录下
cd /usr/local/spark/jars
mkdir kafka
cd ~
cp ./spark-streaming-kafka-0.8_2.11-2.4.0.jar /usr/local/spark/jars/kafka

# 将Kafka安装目录下的libs目录下的所有文件复制到spark的jars目录下
cd /usr/local/kafka/libs
cp ./* /usr/local/spark/jars/kafka   # 进入libs目录后，将当权目录下的所有文件进行拷贝

cd /usr/local/spark/conf
vim spark-env.sh

# kafkaWordCount.py

from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

if __name__ == "__main__":
  if len(sys.argv) != 3:   # 第一个参数默认是self
    print("Usage: kafkaWordCount.py<hostname><port>", file=sys.stderr)
    exit(-1)   # 参数长度不够，自动退出
  sc = SparkContext(appName="pythonStreamingKafkaWordCount")
  ssc = StreamingContext(sc, 1)   # 流计算的指挥官

  zkQuorum,topic = sys.argv[1:]
	kvs = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic:1})   # 建立数据源

  lines = kvs.map(lambda x:x[1])
  counts = lines.flatMap(lambda line:line.split(" ").map(lambda word:(word,1)).reduceByKey(lambda a,b: a+b)    # 第二个 map 函数的作用是形成键值对，因为 reduceByKeyd 的参数必须是键值对

  counts.pprint()
  ssc.start()
  ssc.awaitTermination()

cd /usr/local/spark/mycode/streaming/kafka
/usr/local/spark/bin/spark-submit ./kafkaWordCount.py locaohost:2181 wordsendertest   # 2181是ZK服务器的地址，wordsendertest是topic的名称