使用pandas_udf将火花结构化的数据框架转换为熊猫

Question

我需要将CSV文件读为流，然后将其转换为pandas dataframe。

这是我到目前为止所做的

    DataShema = StructType([ StructField("TimeStamp", LongType(), True), 
    StructField("Count", IntegerType(), True), 
    StructField("Reading", FloatType(), True) ])
    group_columns = ['TimeStamp','Count','Reading']
    @pandas_udf(DataShema, PandasUDFType.GROUPED_MAP)
    def get_pdf(pdf):
        return pd.DataFrame([pdf[group_columns]],columns=[group_columns])
    # getting Surge data from the files
    SrgDF = spark 
        .readStream 
        .schema(DataShema) 
        .csv("ProcessdedData/SurgeAcc")
    mydf = SrgDF.groupby(group_columns).apply(get_pdf)
    qrySrg = SrgDF 
        .writeStream 
        .format("console") 
        .start() 
        .awaitTermination()

我相信，从另一个来源(将Spark结构流数据框架转换为PANDAS DataFrame(，将结构化的流数据框转换为Pandas是不可能的，并且似乎Pandas_udf是正确的方法，但无法确切地发现如何实现这一目标。我需要熊猫数据框来传递我的功能。

edit

当我运行代码(将查询更改为mydf而不是SrgDF(时，我会收到以下错误：pyspark.sql.utils.StreamingQueryException: 'Writing job aborted.n=== Streaming Query ===nIdentifier: [id = 18a15e9e-9762-4464-b6d1-cb2db8d0ac41, runId = e3da131e-00d1-4fed-82fc-65bf377c3f99]nCurrent Committed Offsets: {}nCurrent Available Offsets: {FileStreamSource[file:/home/mls5/Work_Research/Codes/Misc/Python/MachineLearning_ArtificialIntelligence/00_Examples/01_ApacheSpark/01_ComfortApp/ProcessdedData/SurgeAcc]: {"logOffset":0}}nnCurrent State: ACTIVEnThread State: RUNNABLEnnLogical Plan:nFlatMapGroupsInPandas [Count#1], get_pdf(TimeStamp#0L, Count#1, Reading#2), [TimeStamp#10L, Count#11, Reading#12]n+- Project [Count#1, TimeStamp#0L, Count#1, Reading#2]n +- StreamingExecutionRelation FileStreamSource[file:/home/mls5/Work_Research/Codes/Misc/Python/MachineLearning_ArtificialIntelligence/00_Examples/01_ApacheSpark/01_ComfortApp/ProcessdedData/SurgeAcc], [TimeStamp#0L, Count#1, Reading#2]n' 19/05/20 18:32:29 ERROR ReceiverTracker: Deregistered receiver for stream 0: Stopped by driver /usr/local/lib/python3.6/dist-packages/pyarrow/__init__.py:152: UserWarning: pyarrow.open_stream is deprecated, please use pyarrow.ipc.open_stream warnings.warn("pyarrow.open_stream is deprecated, please use "。

edit-2

这是重现错误

的代码

import sys
from pyspark import SparkContext
from pyspark.sql import Row, SparkSession, SQLContext
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
from pyspark.streaming import StreamingContext
from pyspark.sql.types import *
import pandas as pd
from pyspark.sql.functions import pandas_udf, PandasUDFType
import pyarrow as pa
import glob
#####################################################################################
if __name__ == '__main__' :
    spark = SparkSession 
        .builder 
        .appName("RealTimeIMUAnalysis") 
        .getOrCreate() 
    spark.conf.set("spark.sql.execution.arrow.enabled", "true")
    # reduce verbosity
    sc = spark.sparkContext
    sc.setLogLevel("ERROR")
    ##############################################################################
    # using the saved files to do the Analysis
    DataShema = StructType([ StructField("TimeStamp", LongType(), True), 
    StructField("Count", IntegerType(), True), 
    StructField("Reading", FloatType(), True) ])
    group_columns = ['TimeStamp','Count','Reading']
    @pandas_udf(DataShema, PandasUDFType.GROUPED_MAP)
    def get_pdf(pdf):
        return pd.DataFrame([pdf[group_columns]],columns=[group_columns])
    # getting Surge data from the files
    SrgDF = spark 
        .readStream 
        .schema(DataShema) 
        .csv("SurgeAcc")
    mydf = SrgDF.groupby('Count').apply(get_pdf)
    #print(mydf)
    qrySrg = mydf 
        .writeStream 
        .format("console") 
        .start() 
        .awaitTermination()

要运行，您需要创建一个名为 SurgeAcc的文件夹，并在内部使用以下格式创建一个CSV文件：

TimeStamp,Count,Reading
1557011317299,45148,-0.015494
1557011317299,45153,-0.015963
1557011319511,45201,-0.015494
1557011319511,45221,-0.015494
1557011315134,45092,-0.015494
1557011315135,45107,-0.014085
1557011317299,45158,-0.015963
1557011317299,45163,-0.015494
1557011317299,45168,-0.015024`

Answer 1

您的返回pandas_udf dataframe与指定的模式不匹配。

请注意，pandas_udf的输入将是pandas dataframe，并且还返回pandas dataframe。

您可以使用pandas_udf内的所有熊猫功能。唯一要确保的是returndatashema应该与功能的实际输出匹配。

ReturnDataShema = StructType([StructField("TimeStamp", LongType(), True), 
                            StructField("Count", IntegerType(), True), 
                            StructField("Reading", FloatType(), True), 
                            StructField("TotalCount", FloatType(), True)])
@pandas_udf(ReturnDataShema, PandasUDFType.GROUPED_MAP)
    def get_pdf(pdf):
        # This following stmt is causing schema mismatch
        # return pd.DataFrame([pdf[group_columns]],columns=[group_columns])
        # If you want to return all the rows of pandas dataframe
        # you can simply
        # return pdf
        # If you want to do any aggregations, you can do like the below, or use pandas query
        # but make sure the return pandas dataframe complies with ReturnDataShema
        total_count = pdf['Count'].sum()
        return pd.DataFrame([(pdf.TimeStamp[0],pdf.Count[0],pdf.Reading[0],total_count)])