apache-spark - 如何使用 spark 数据框评估 spark Dstream 对象

Question

我正在编写一个 spark 应用程序，我需要根据历史数据评估流数据，该数据位于 sql server 数据库中

现在的想法是，spark 将从数据库中获取历史数据并将其保存在内存中，并根据它评估流数据。

现在我正在获取流数据

import re
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext,functions as func,Row


sc = SparkContext("local[2]", "realtimeApp")
ssc = StreamingContext(sc,10)
files = ssc.textFileStream("hdfs://RealTimeInputFolder/")

########Lets get the data from the db which is relavant for streaming ###

driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
dataurl = "jdbc:sqlserver://myserver:1433"
db = "mydb"
table = "stream_helper"
credential = "my_credentials"

########basic data for evaluation purpose ########



files_count = files.flatMap(lambda file: file.split( ))

pattern =  '(TranAmount=Decimal.{2})(.[0-9]*.[0-9]*)(\\S+ )(TranDescription=u.)([a-zA-z\\s]+)([\\S\\s]+ )(dSc=u.)([A-Z]{2}.[0-9]+)'


tranfiles = "wasb://myserver.blob.core.windows.net/RealTimeInputFolder01/"

def getSqlContextInstance(sparkContext):
    if ('sqlContextSingletonInstance' not in globals()):
        globals()['sqlContextSingletonInstance'] = SQLContext(sparkContext)
    return globals()['sqlContextSingletonInstance']


def pre_parse(logline):
    """
    to read files as rows of sql in pyspark streaming using the pattern . for use of logging 
    added 0,1 in case there is any failure in processing by this pattern

    """
    match = re.search(pattern,logline)
    if match is None:
        return(line,0)
    else:
        return(
        Row(
        customer_id = match.group(8)
        trantype = match.group(5)
        amount = float(match.group(2))
        ),1)


def parse():
    """
    actual processing is happening  here 
    """
    parsed_tran = ssc.textFileStream(tranfiles).map(preparse)
    success = parsed_tran.filter(lambda s: s[1] == 1).map(lambda x:x[0])
    fail = parsed_tran.filter(lambda s:s[1] == 0).map(lambda x:x[0])
    if fail.count() > 0:
        print "no of non parsed file : %d", % fail.count()

    return success,fail

success ,fail = parse()

现在我想通过我从历史数据中得到的数据框来评估它

base_data = sqlContext.read.format("jdbc").options(driver=driver,url=dataurl,database=db,user=credential,password=credential,dbtable=table).load()

现在，既然这是作为数据框返回的，我如何将其用于我的目的。
流媒体节目指南here说
“您必须使用 StreamingContext 正在使用的 SparkContext 创建一个 SQLContext。”

现在这让我更加困惑如何将现有数据帧与流对象一起使用。任何帮助表示高度赞赏。

Answer 1

要操作 DataFrames，你总是需要一个 SQLContext 所以你可以实例化它:

sc = SparkContext("local[2]", "realtimeApp")
sqlc = SQLContext(sc)
ssc = StreamingContext(sc, 10)

这 2 个上下文( SQLContext 和 StreamingContext )将共存于同一个作业中，因为它们与相同的 相关联SparkContext .
但是，请记住，您不能在同一个作业中实例化两个不同的 SparkContext。

从 DStream 创建 DataFrame 后，您可以将历史 DataFrame 与从流创建的 DataFrame 连接起来。
为此，我会做类似的事情:

yourDStream.foreachRDD(lambda rdd: sqlContext
    .createDataFrame(rdd)
    .join(historicalDF, ...)
    ...
)

考虑在操作流时需要用于连接的流数据量，您可能对 windowed functions 感兴趣。