Spark & PySpark 使用手册

栏目: 编程工具 · 发布时间: 5年前

内容简介:最近公司有一个安全方面的业务,需要实时监控所有访客的情况。之前是定时去查询这类方法仅仅是定义逻辑,并不会立即执行,即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()	# 等待终止信号

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

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

The Art of Computer Programming, Volume 2

The Art of Computer Programming, Volume 2

Knuth, Donald E. / Addison-Wesley Professional / 1997-11-04 / USD 79.99

Finally, after a wait of more than thirty-five years, the first part of Volume 4 is at last ready for publication. Check out the boxed set that brings together Volumes 1 - 4A in one elegant case, and ......一起来看看 《The Art of Computer Programming, Volume 2》 这本书的介绍吧!

在线进制转换器
在线进制转换器

各进制数互转换器

图片转BASE64编码
图片转BASE64编码

在线图片转Base64编码工具

Base64 编码/解码
Base64 编码/解码

Base64 编码/解码