Spark & PySpark 使用手册

内容简介：最近公司有一个安全方面的业务，需要实时监控所有访客的情况。之前是定时去查询这类方法仅仅是定义逻辑，并不会立即执行，即lazy特性。目的是将一个RDD转为新的RDD。不会产生新的RDD，而是直接运行，得到我们想要的结果。

最近公司有一个安全方面的业务，需要实时监控所有访客的情况。之前是定时去查询 Elasticsearch 的接口进行统计分析，但是这个时间间隔不好把握，并且 Elasticsearch 并不适合特别实时的查询操作。实时的分布式流计算引擎首推 Spark ，它与 Hadoop 等相比的优势在这里讲得比较清楚了。

try...catch...
pip install pyspark

基本运算

下面是所有运算方法的集合，其中有些方法仅用于键值对，有些方法仅用于数据流

Transformation(转换)

这类方法仅仅是定义逻辑，并不会立即执行，即lazy特性。目的是将一个RDD转为新的RDD。

datas.map(lambda a, (a, 1))

Action(执行)

不会产生新的RDD，而是直接运行，得到我们想要的结果。

collect(): 以数组的形式，返回数据集中所有的元素
count(): 返回数据集中元素的个数
take(n): 返回数据集的前N个元素
takeOrdered(n): 升序排列，取出前N个元素
takeOrdered(n, lambda x: -x): 降序排列，取出前N个元素
first(): 返回数据集的第一个元素
min(): 取出最小值
max(): 取出最大值
stdev(): 计算标准差
sum(): 求和
mean(): 平均值
countByKey(): 统计各个key值对应的数据的条数
lookup(key): 根据传入的key值来查找对应的Value值
foreach(func): 对集合中每个元素应用func

Persistence(持久化)

persist(): 将数据按默认的方式进行持久化
unpersist(): 取消持久化
saveAsTextFile(path): 将数据集保存至文件

应用场景

创建简单的RDD

from pyspark.sql import SparkConf, SparkContext
rdd = sc.parallelize(['abc', def'])	// 直接创建rdd

读取CSV文件

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("test") \
    .config("spark.some.config.option", "一些设置") \
    .getOrCreate()

df = spark.read.csv("/home/Users/haofly/test.csv", header=True, sep="|")	# 读取文件
print(df.collect())

Spark Streaming流计算

Spark Streaming
StreamingContext

从文件流读取数据

conf = SparkConf().setAppName("test").setMaster("local[2]")	# 表示运行在本地模式，并且启动2个工作线程
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 30)	# 每隔10秒钟自动进行一次流计算

lines = ssc.textFileStream('file:///Users/haofly/log')
words = lines.map(lambda line: line.strip())
words.pprint()
ssc.start()
ssc.awaitTermination()

从kafka读取数据

首先得从 maven仓库下载对应的版本，注意这里需要下载 assembly 的包，这里的2.11是scala的版本，2.4.3是pyspark的版本号，也是spark的版号，如果下载后的包不能用，那就尝试换一个版本吧。可以通过这篇文章搭建测试用的kafka集群

# 指定spark-streaming-kafka的jar包
os.environ[
    "PYSPARK_SUBMIT_ARGS"
] = "--jars /test/jars/kafka/spark-streaming-kafka-0-8-assembly_2.11-2.4.3.jar pyspark-shell"

conf = SparkConf().setAppName("test").setMaster("local[2]")	# 表示运行在本地模式，并且启动2个工作线程
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 10)	# 每隔10秒钟自动进行一次流计算

zkQuorum, topic = "zookeeper:2181", "test"
kvs = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic: 1})
lines = kvs.map(lambda x: x[1])

def myadd(a, b):	# 只能在传入的函数中捕获异常
  try:
	  return a+b
  catch:
    pass	# tolog

def myadd_inv(a, b):
  return a-b

rdd = lines.map(lambda x: (x, 1)).reduceByKeyAndWIndow(myadd, myadd_inv, 60)	# 统计时间窗口60秒内的数据
rdd.pprint()	# 每次统计都打印rdd的数据

ssc.start()		# 异步执行
ssc.awaitTermination()	# 等待终止信号

以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持码农网