pyspark.streaming module
Module contents
class pyspark.streaming.StreamingContext(sparkContext, batchDuration=None, jssc=None)
Bases: object
Spark Streaming功能的主要入口點(diǎn)。 StreamingContext表示與Spark集群的連接，可用于創(chuàng)建DStream各種輸入源。 它可以來自現(xiàn)有的SparkContext。 在創(chuàng)建和轉(zhuǎn)換DStream之后，可以分別使用context.start（）和context.stop（）來啟動(dòng)和停止流計(jì)算。 context.awaitTermination（）允許當(dāng)前線程通過stop（）或異常等待上下文的終止。
- addStreamingListener(streamingListener)
添加[[org.apache.spark.streaming.scheduler.StreamingListener]]對(duì)象來接收與流式傳輸相關(guān)的系統(tǒng)事件。 - awaitTermination(timeout=None)
等待執(zhí)行停止。 - awaitTerminationOrTimeout(timeout)
等待執(zhí)行停止。 如果停止或在執(zhí)行過程中拋出錯(cuò)誤，則返回true; 或者如果從方法返回之前超過了等待時(shí)間，則返回false。 - binaryRecordsStream(directory, recordLength)
創(chuàng)建一個(gè)輸入流，用于監(jiān)視Hadoop兼容的文件系統(tǒng)是否存在新文件，并將其作為具有固定長度記錄的平面二進(jìn)制文件進(jìn)行讀取。 必須通過從同一個(gè)文件系統(tǒng)中的另一個(gè)位置“移動(dòng)”文件來將文件寫入受監(jiān)視的目錄。 文件名稱是.的將被忽略。 - checkpoint(directory)
設(shè)置上下文以定期檢查DStream操作是否具有主容錯(cuò)功能。 該圖會(huì)檢查每個(gè)批處理間隔。 - classmethod getActive()
返回當(dāng)前活動(dòng)的StreamingContext（即，如果存在已啟動(dòng)但未停止的上下文）或None。 - classmethod getActiveOrCreate(checkpointPath, setupFunc)
返回活動(dòng)的StreamingContext（即，當(dāng)前已啟動(dòng)但未停止），或者從檢查點(diǎn)數(shù)據(jù)重新創(chuàng)建StreamingContext，或使用提供的setupFunc函數(shù)創(chuàng)建新的StreamingContext。 如果checkpointPath為None或不包含有效的檢查點(diǎn)數(shù)據(jù)，則將調(diào)用setupFunc來創(chuàng)建新的上下文并設(shè)置DStreams。 - classmethod getOrCreate(checkpointPath, setupFunc)
重新從檢查點(diǎn)中的數(shù)據(jù)創(chuàng)建一個(gè)StreamingContext或創(chuàng)建一個(gè)新的StreamingContext。 如果提供的checkpointPath中存在檢查點(diǎn)數(shù)據(jù)，則將從檢查點(diǎn)數(shù)據(jù)重新創(chuàng)建StreamingContext。 如果數(shù)據(jù)不存在，則提供的setupFunc將用于創(chuàng)建新的上下文。 - queueStream(rdds, oneAtATime=True, default=None)
從RDD的隊(duì)列或列表創(chuàng)建一個(gè)輸入流。 在每個(gè)批次中，它將處理隊(duì)列返回的一個(gè)或全部RDD。 - remember(duration)
在此上下文中設(shè)置每個(gè)DStream，以記住在上一個(gè)給定的持續(xù)時(shí)間內(nèi)生成的RDD。 DStream只記錄在有限的時(shí)間內(nèi)和垃圾回收機(jī)制釋放它之前的RDD。 此方法允許開發(fā)人員指定如何記住RDD（如果開發(fā)人員希望查詢DStream計(jì)算之外的舊數(shù)據(jù)）。 - socketTextStream(hostname, port, storageLevel=StorageLevel(True, True, False, False, 2))
從TCP源hostname:port創(chuàng)建一個(gè)輸入。 使用TCP套接字接收數(shù)據(jù)，接收字節(jié)被解釋為UTF8編碼的，\n分隔的。 - sparkContext
返回與此StreamingContext關(guān)聯(lián)的SparkContext。 - start()
開始執(zhí)行流。 - stop(stopSparkContext=True, stopGraceFully=False)
停止流的執(zhí)行，并確保所有接收到的數(shù)據(jù)已被處理。
Parameters:
- stopSparkContext – 停止關(guān)聯(lián)的SparkContext或不。
- stopGracefully – 等待所有接收到的數(shù)據(jù)處理完成之后優(yōu)雅地停止。
- textFileStream(directory)
創(chuàng)建一個(gè)輸入流，用于監(jiān)視與Hadoop兼容的文件系統(tǒng)中的新文件，并將其作為文本文件讀取。 文件必須通過從同一個(gè)文件系統(tǒng)中的另一個(gè)位置“移動(dòng)”來監(jiān)視目錄。 文件名稱是.的將被忽略。 - transform(dstreams, transformFunc)
通過對(duì)源 DStream 的每個(gè) RDD 應(yīng)用 RDD-to-RDD 函數(shù)，創(chuàng)建一個(gè)新的 DStream. 這個(gè)可以在 DStream 中的任何 RDD 操作中使用. - union(*dstreams)
從相同類型和相同的滑動(dòng)時(shí)間的多個(gè)DStream中創(chuàng)建一個(gè)統(tǒng)一的DStream。
class pyspark.streaming.DStream(jdstream, ssc, jrdd_deserializer)
Bases: object
離散化流（DStream）是Spark Streaming中的基本抽象，它是連續(xù)的（相同類型的）RDD序列，用于表示連續(xù)的數(shù)據(jù)流（有關(guān)RDD的更多詳細(xì)信息，請(qǐng)參閱Spark核心文檔中的RDD）。
可以使用StreamingContext從實(shí)時(shí)數(shù)據(jù)（例如來自TCP套接字，Kafka，F(xiàn)lume等的數(shù)據(jù)）創(chuàng)建DStream，或者可以通過使用諸如map，window和reduceByKeyAndWindow之類的操作來轉(zhuǎn)換現(xiàn)有DStream來生成DStream。 當(dāng)Spark Streaming程序正在運(yùn)行時(shí)，每個(gè)DStream都會(huì)定期生成一個(gè)RDD，可以通過實(shí)時(shí)數(shù)據(jù)或通過轉(zhuǎn)換父DStream生成的RDD來生成RDD。
DStreams的內(nèi)部特點(diǎn)是一些基本屬性：
- DStream依賴其他DStream列表
- DStream生成RDD的時(shí)間間隔
- 在每個(gè)時(shí)間間隔之后用于生成RDD的函數(shù)
- cache()
使用默認(rèn)存儲(chǔ)級(jí)別（MEMORY_ONLY）存儲(chǔ)此DStream的RDD。 - checkpoint(interval)
啟用此DStream的RDD的定期檢查點(diǎn)。 - cogroup(other, numPartitions=None)
通過在此DStream的RDD和其他DStream之間應(yīng)用“cogroup”來返回一個(gè)新的DStream。
散列分區(qū)用于生成具有numPartitions分區(qū)的RDD。 - combineByKey(createCombiner, mergeValue, mergeCombiners, numPartitions=None)
通過將combineByKey應(yīng)用于每個(gè)RDD來返回一個(gè)新的DStream。 - context()
返回與此DStream關(guān)聯(lián)的StreamingContext - count()
返回一個(gè)新的DStream，其中每個(gè)RDD具有通過對(duì)該DStream的每個(gè)RDD進(jìn)行計(jì)數(shù)而生成的單個(gè)元素。 - countByValue()
返回一個(gè)新的DStream，其中每個(gè)RDD包含此DStream的每個(gè)RDD中每個(gè)不同值的計(jì)數(shù)。 - countByValueAndWindow(windowDuration, slideDuration, numPartitions=None)
返回一個(gè)新的DStream，其中每個(gè)RDD包含在該DStream上的滑動(dòng)窗口中的RDD中的不同元素的計(jì)數(shù)。
Parameters:
- windowDuration - 窗戶的寬度; 必須是DStream批處理間隔的倍數(shù)
- slideDuration - 窗口的滑動(dòng)間隔;必須是DStream批處理間隔的倍數(shù)
- numPartitions - 新DStream中每個(gè)RDD的分區(qū)數(shù)量
- countByWindow(windowDuration, slideDuration)
返回一個(gè)新的DStream，其中每個(gè)RDD都有一個(gè)通過計(jì)算該DStream中窗口中元素?cái)?shù)目而生成的單個(gè)元素。 windowDuration和slideDuration是在window（）操作中定義的。
這相當(dāng)于window（windowDuration，slideDuration）.count（），但是如果window很大，效率會(huì)更高。 - filter(f)
返回僅包含滿足f的元素的新DStream。 - flatMap(f, preservesPartitioning=False)
通過對(duì)該DStream的所有元素應(yīng)用一個(gè)函數(shù)來返回一個(gè)新的DStream，然后展平結(jié)果。 - flatMapValues(f)
通過將flatmap函數(shù)應(yīng)用于此DStream中每個(gè)鍵值對(duì)中的value而不更改key來返回新的DStream。 - foreachRDD(func)
在此DStream中的每個(gè)RDD上應(yīng)用一個(gè)函數(shù)。 - fullOuterJoin(other, numPartitions=None)
通過在此DStream的RDD和其他DStream之間應(yīng)用“完全外連接”來返回一個(gè)新的DStream。
Hash分區(qū)用于生成具有numPartitions分區(qū)的RDD。 - glom()
返回一個(gè)新的DStream，其中通過將glom（）應(yīng)用于此DStream的RDD來生成RDD。 - groupByKey(numPartitions=None)
通過在每個(gè)RDD上應(yīng)用groupByKey來返回一個(gè)新的DStream。 - groupByKeyAndWindow(windowDuration, slideDuration, numPartitions=None)
通過在滑動(dòng)窗口上應(yīng)用groupByKey來返回一個(gè)新的DStream。 與DStream.groupByKey（）類似，但將其應(yīng)用于滑動(dòng)窗口。 - join(other, numPartitions=None)
通過應(yīng)用此DStream的RDD和其他DStream之間的“join”來返回一個(gè)新的DStream。
Hash分區(qū)用于生成具有numPartitions分區(qū)的RDD。 - leftOuterJoin(other, numPartitions=None)
通過應(yīng)用此DStream的RDD和其他DStream之間的“l(fā)eft outer join”來返回一個(gè)新的DStream。
Hash分區(qū)用于生成具有numPartitions分區(qū)的RDD。 - map(f, preservesPartitioning=False)
通過對(duì)DStream的每個(gè)元素應(yīng)用一個(gè)函數(shù)來返回一個(gè)新的DStream。 - mapPartitions(f, preservesPartitioning=False)
返回一個(gè)新的DStream，其中通過將mapPartitions（）應(yīng)用于此DStream的每個(gè)RDD來生成每個(gè)RDD。 - mapPartitionsWithIndex(f, preservesPartitioning=False)
通過將mapPartitionsWithIndex（）應(yīng)用到此DStream的每個(gè)RDD，返回一個(gè)新的DStream，其中每個(gè)RDD都是通過應(yīng)用mapPartitionsWithIndex（）生成的。 - mapValues(f)
通過將映射函數(shù)應(yīng)用于此DStream中每個(gè)鍵值對(duì)的值，而不更改鍵，可以返回一個(gè)新的DStream。 - partitionBy(numPartitions, partitionFunc=<function portable_hash at 0x7f51f1ac0668>)
返回使用指定分區(qū)程序?qū)γ總€(gè)RDD進(jìn)行分區(qū)的DStream副本。 - persist(storageLevel)
以給定的存儲(chǔ)級(jí)別存儲(chǔ)該DStream的RDD。 - pprint(num=10)
打印此DStream中生成的每個(gè)RDD的第一個(gè)num元素。 - reduce(func)
Return a new DStream in which each RDD has a single element generated by reducing each RDD of this DStream. - reduceByKey(func, numPartitions=None)
通過將reduceByKey應(yīng)用于每個(gè)RDD來返回一個(gè)新的DStream。 - reduceByKeyAndWindow(func, invFunc, windowDuration, slideDuration=None, numPartitions=None, filterFunc=None)
通過在滑動(dòng)窗口上應(yīng)用遞增的reduceByKey來返回一個(gè)新的DStream。 - reduceByWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration)
- repartition(numPartitions)
以增加或減少的并行度返回一個(gè)新的DStream。 - rightOuterJoin(other, numPartitions=None)
通過在此DStream的RDD和其他DStream之間應(yīng)用“右外連接”來返回一個(gè)新的DStream。
Hash分區(qū)用于生成具有numPartitions分區(qū)的RDD。 - saveAsTextFiles(prefix, suffix=None)
將這個(gè)DStream中的每個(gè)RDD保存為文本文件，使用字符串表示元素。 - slice(begin, end)
返回“begin”到“end”之間的所有RDD（包括兩者） - transform(func)
返回一個(gè)新的DStream，其中每個(gè)RDD是通過在該DStream的每個(gè)RDD上應(yīng)用一個(gè)函數(shù)而生成的。
func可以有rdd這一個(gè)參數(shù)，或者有兩個(gè)參數(shù)（time，rdd） - transformWith(func, other, keepSerializer=False)
返回一個(gè)新的DStream，其中每個(gè)RDD是通過在該DStream和其他DStream的每個(gè)RDD上應(yīng)用一個(gè)函數(shù)而生成的。
func可以有兩個(gè)參數(shù)（rdd_a，rdd_b）或者有三個(gè)參數(shù)（time，rdd_a，rdd_b） - union(other)
通過使用此DStream來聯(lián)合另一個(gè)DStream的數(shù)據(jù)來返回一個(gè)新的DStream。 - updateStateByKey(updateFunc, numPartitions=None, initialRDD=None)
返回一個(gè)新的“state”DStream，其中通過對(duì)鍵的先前狀態(tài)和鍵的新值應(yīng)用給定函數(shù)來更新每個(gè)鍵的狀態(tài)。 - window(windowDuration, slideDuration=None)
返回一個(gè)新的DStream，其中每個(gè)RDD包含所有在這個(gè)DStream上的滑動(dòng)時(shí)間窗口中看到的元素。
class pyspark.streaming.StreamingListener
Bases: object
class Java
implements = ['org.apache.spark.streaming.api.java.PythonStreamingListener']
- StreamingListener.onBatchCompleted(batchCompleted)
當(dāng)一批作業(yè)處理完成時(shí)調(diào)用。 - StreamingListener.onBatchStarted(batchStarted)
當(dāng)處理一批作業(yè)已經(jīng)開始時(shí)調(diào)用。 - StreamingListener.onBatchSubmitted(batchSubmitted)
當(dāng)一批作業(yè)已經(jīng)提交處理時(shí)調(diào)用。 - StreamingListener.onOutputOperationCompleted(outputOperationCompleted)
一批作業(yè)的處理已經(jīng)完成時(shí)調(diào)用。 - StreamingListener.onOutputOperationStarted(outputOperationStarted)
批量作業(yè)的處理已經(jīng)開始時(shí)調(diào)用。 - StreamingListener.onReceiverError(receiverError)
當(dāng)接收方報(bào)告錯(cuò)誤時(shí)調(diào)用。 - StreamingListener.onReceiverStarted(receiverStarted)
接收者啟動(dòng)時(shí)調(diào)用。 - StreamingListener.onReceiverStopped(receiverStopped)
接收者停止時(shí)調(diào)用。
pyspark.streaming.kafka module
class pyspark.streaming.kafka.Broker(host, port)
Bases: object
表示kafka broker的主機(jī)和端口信息。
class pyspark.streaming.kafka.KafkaMessageAndMetadata(topic, partition, offset, key, message)
Bases: object
kafka消息和元數(shù)據(jù)信息。 包括topic, partition, offset and message.
- key
- message
class pyspark.streaming.kafka.KafkaUtils
Bases: object
- static createDirectStream(ssc, topics, kafkaParams, fromOffsets=None, keyDecoder=<function utf8_decoder at 0x7f51e9a60b90>, valueDecoder=<function utf8_decoder at 0x7f51e9a60b90>, messageHandler=None)
- Note: 實(shí)驗(yàn)階段
創(chuàng)建一個(gè)直接從Kafka broker和特定偏移量中獲取消息的輸入流。
這不是一個(gè)基于接收者的Kafka輸入流，它在每個(gè)批次持續(xù)時(shí)間直接從Kafka拉取消息，并且不經(jīng)過存儲(chǔ)處理。
這不使用Zookeeper來存儲(chǔ)偏移量。 消耗的偏移量由流自身跟蹤。 為了與依賴于Zookeeper的Kafka監(jiān)控工具進(jìn)行互操作，您必須自己從流應(yīng)用程序更新Kafka / Zookeeper。 您可以從生成的RDD中訪問每個(gè)批次中使用的偏移量。
要從驅(qū)動(dòng)程序失敗中恢復(fù)，必須在StreamingContext中啟用檢查點(diǎn)。 關(guān)于消耗偏移量的信息可以從檢查點(diǎn)恢復(fù)。
Parameters:
- ssc – StreamingContext object.
- topics – list of topic_name to consume.
- kafkaParams – Kafka的附加參數(shù)
- fromOffsets – 每個(gè)主題/分區(qū)Kafka偏移量定義（包括）流的起始點(diǎn)。
- keyDecoder – A function used to decode key (default is utf8_decoder).
- valueDecoder – A function used to decode value (default is utf8_decoder).
- messageHandler – 用于轉(zhuǎn)換KafkaMessageAndMetadata的函數(shù)。 你可以使用messageHandler來評(píng)估m(xù)eta（默認(rèn)是None）。
Returns: A DStream object
- static createRDD(sc, kafkaParams, offsetRanges, leaders=None, keyDecoder=<function utf8_decoder at 0x7f51e9a60b90>, valueDecoder=<function utf8_decoder at 0x7f51e9a60b90>, messageHandler=None)
使用每個(gè)主題和分區(qū)的偏移量范圍，從Kafka創(chuàng)建一個(gè)RDD。
Parameters:
- sc – SparkContext object
- kafkaParams – Additional params for Kafka
- offsetRanges – offsetRange列表來指定topic：partition：[start，end）來消費(fèi)
- leaders – Kafka brokers for each TopicAndPartition in offsetRanges. May be an empty map, in which case leaders will be looked up on the driver.
- keyDecoder – A function used to decode key (default is utf8_decoder)
- valueDecoder – A function used to decode value (default is utf8_decoder)
- messageHandler – A function used to convert KafkaMessageAndMetadata. You can assess meta using messageHandler (default is None).
Returns: A DStream object
- static createStream(ssc, zkQuorum, groupId, topics, kafkaParams=None, storageLevel=StorageLevel(True, True, False, False, 2), keyDecoder=<function utf8_decoder at 0x7f51e9a60b90>, valueDecoder=<function utf8_decoder at 0x7f51e9a60b90>)
創(chuàng)建一個(gè)從Kafka Broker中提取消息的輸入流。
Parameters:
- ssc – StreamingContext object
- zkQuorum – Zookeeper quorum (hostname:port,hostname:port,..).
- groupId – The group id for this consumer.
- topics – Dict of (topic_name -> numPartitions) to consume. Each partition is consumed in its own thread.
- kafkaParams – Additional params for Kafka
- storageLevel – RDD storage level.
- keyDecoder – A function used to decode key (default is utf8_decoder)
- valueDecoder – A function used to decode value (default is utf8_decoder)
Returns: A DStream object
class pyspark.streaming.kafka.OffsetRange(topic, partition, fromOffset, untilOffset)
Bases: object
表示來自單個(gè)kafka TopicAndPartition的一系列偏移量。
class pyspark.streaming.kafka.TopicAndPartition(topic, partition)
Bases: object
代表Kafka的特定主題和分區(qū)。
pyspark.streaming.kafka.utf8_decoder(s)
將Unicode解碼為UTF-8
pyspark.streaming.kinesis module
class pyspark.streaming.kinesis.KinesisUtils
Bases: object
- static createStream(ssc, kinesisAppName, streamName, endpointUrl, regionName, initialPositionInStream, checkpointInterval, storageLevel=StorageLevel(True, True, False, False, 2), awsAccessKeyId=None, awsSecretKey=None, decoder=<function utf8_decoder at 0x7f51e9541d70>, stsAssumeRoleArn=None, stsSessionName=None, stsExternalId=None)
創(chuàng)建一個(gè)從Kinesis流中獲取消息的輸入流。 這使用Kinesis客戶端庫（KCL）從Kinesis中獲取消息。
Parameters:
- ssc – StreamingContext object
- kinesisAppName – Kinesis application name used by the Kinesis Client Library (KCL) to update DynamoDB
- streamName – Kinesis stream name
- endpointUrl – Url of Kinesis service (e.g., https://kinesis.us-east-1.amazonaws.com)
- regionName – Name of region used by the Kinesis Client Library (KCL) to update DynamoDB (lease coordination and checkpointing) and CloudWatch (metrics)
- initialPositionInStream – In the absence of Kinesis checkpoint info, this is the worker’s initial starting position in the stream. The values are either the beginning of the stream per Kinesis’ limit of 24 hours (InitialPositionInStream.TRIM_HORIZON) or the tip of the stream (InitialPositionInStream.LATEST).
- checkpointInterval – Checkpoint interval for Kinesis checkpointing. See the Kinesis Spark Streaming documentation for more details on the different types of checkpoints.
- storageLevel – Storage level to use for storing the received objects (default is StorageLevel.MEMORY_AND_DISK_2)
- awsAccessKeyId – AWS AccessKeyId (default is None. If None, will use DefaultAWSCredentialsProviderChain)
- awsSecretKey – AWS SecretKey (default is None. If None, will use DefaultAWSCredentialsProviderChain)
- decoder – A function used to decode value (default is utf8_decoder)
- stsAssumeRoleArn – ARN of IAM role to assume when using STS sessions to read from the Kinesis stream (default is None).
- stsSessionName – Name to uniquely identify STS sessions used to read from Kinesis stream, if STS is being used (default is None).
- stsExternalId – External ID that can be used to validate against the assumed IAM role’s trust policy, if STS is being used (default is None).
Returns: A DStream object
class pyspark.streaming.kinesis.InitialPositionInStream
Bases: object
LATEST = 0
TRIM_HORIZON = 1
pyspark.streaming.kinesis.utf8_decoder(s)
將Unicode解碼為UTF-8
pyspark.streaming.flume.module
class pyspark.streaming.flume.FlumeUtils
Bases: object
- static createPollingStream(ssc, addresses, storageLevel=StorageLevel(True, True, False, False, 2), maxBatchSize=1000, parallelism=5, bodyDecoder=<function utf8_decoder at 0x7f51eb4416e0>)
創(chuàng)建要與Flume代理上部署的Spark agent一起使用的輸入流。 這個(gè)流將輪詢接收器的數(shù)據(jù)，并將提供事件，因?yàn)樗鼈兪强捎玫摹?br> Parameters:
- ssc – StreamingContext object
- addresses – 運(yùn)行Spark Sink的（主機(jī)，端口）列表
- storageLevel –用于存儲(chǔ)接收對(duì)象的存儲(chǔ)級(jí)別
- maxBatchSize – 在單個(gè)RPC調(diào)用中從Spark接收器中拉取的最大事件數(shù)量
- parallelism – 此流應(yīng)發(fā)送到接收器的并發(fā)請(qǐng)求數(shù)。 請(qǐng)注意，同時(shí)拉取更多請(qǐng)求將導(dǎo)致此流使用更多的線程
- bodyDecoder – A function used to decode body (default is utf8_decoder)
Returns: A DStream object
- static createStream(ssc, hostname, port, storageLevel=StorageLevel(True, True, False, False, 2), enableDecompression=False, bodyDecoder=<function utf8_decoder at 0x7f51eb4416e0>)
創(chuàng)建一個(gè)從Flume中提取事件的輸入流。
Parameters:
- ssc – StreamingContext object
- hostname – Hostname of the slave machine to which the flume data will be sent
- port – Port of the slave machine to which the flume data will be sent
- storageLevel – Storage level to use for storing the received objects
- enableDecompression – Should netty server decompress input stream
- bodyDecoder – A function used to decode body (default is utf8_decoder)
Returns: A DStream object
pyspark.streaming.flume.utf8_decoder(s)
將Unicode解碼為UTF-8
