如何使用 PySpark 从日常文件中加载滚动窗口？

Question

我有大量相当大的日常文件存储在博客存储引擎（S3、Azure datalake exc.. exc..）data1900-01-01.csv, data1900-01-02.csv,....,data2017-04-27.csv。我的目标是执行滚动 N 天线性回归，但我在数据加载方面遇到了麻烦。如果没有嵌套 RDD's.，我不确定如何做到这一点 每个.csv 文件的架构都是相同的。

换句话说，对于每个日期d_t，我需要数据x_t 并加入数据(x_t-1, x_t-2,... x_t-N)。

如何使用 PySpark 加载这些日常文件的 N 天窗口？我能找到的所有 PySpark 示例似乎都是从一个非常大的文件或数据集加载的。

这是我当前代码的示例：

dates = [('1995-01-03', '1995-01-04', '1995-01-05'), ('1995-01-04', '1995-01-05', '1995-01-06')]

p = sc.parallelize(dates)
def test_run(date_range):
    dt0 = date_range[-1] #get the latest date
    s = '/daily/data{}.csv'
    df0 = spark.read.csv(s.format(dt0), header=True, mode='DROPMALFORM')
    file_list = [s.format(dt) for dt in date_range[:-1]] # Get a window of trailing dates
    df1 = spark.read.csv(file_list, header=True, mode='DROPMALFORM')
    return 1

p.filter(test_run) 

p.map(test_run) #fails with same error as p.filter

我正在使用 PySpark 版本 '2.1.0'

我在 Azure HDInsight 集群 jupyter notebook 上运行它。

spark 这里是<class 'pyspark.sql.session.SparkSession'> 类型

一个更小更可重现的例子如下：

p = sc.parallelize([1, 2, 3])
def foo(date_range):
    df = spark.createDataFrame([(1, 0, 3)], ["a", "b", "c"])
    return 1
p.filter(foo).count()

Answer 1

您最好使用Dataframes 而不是RDD。 Dataframe 的 read.csv api 接受路径列表，例如 -

pathList = ['/path/to/data1900-01-01.csv','/path/to/data1900-01-02.csv']
df = spark.read.csv(pathList)

have a look at documentation for read.csv

您可以通过在 N 天的窗口上进行一些日期操作来形成数据文件的路径列表，例如 "path/to/data"+datetime.today().strftime("%Y-%m-%d"))+.csv" （这只会让您获得今天的文件名，但不难计算出日期N天）

但是请记住，所有日期 csv 的架构都应该相同才能使上述工作。

编辑：当您并行化日期列表（即p）时，每个日期都由不同的执行程序单独处理，因此 test_run2 的输入并不是真正的日期列表，它是一个单独的字符串，例如 1995-01-01

试试这个，看看这是否有效。

# Get the list of dates 
date_range = window(dates, N) 
s = '/daily/data{}.csv'

dt0 = date_range[-1] # most recent file
df0 = spark.read.csv(s.format(dt0), header=True, mode='DROPMALFORM') 

# read previous files
file_list = [s.format(dt) for dt in date_range[:-1]]
df1 = spark.read.csv(file_list, header=True, mode='DROPMALFORM')

r, resid = computeLinearRegression(df0,df1)
r.write.save('daily/r{}.csv'.format(dt0))
resid.write.save('/daily/resid{}.csv'.format(dt0))