如何将变量传递给 Pyspark 序列以生成时间序列？

Question

我想生成一个时间序列，从 2021-12-01 到 2021-12-31，但我想将带有变量的值传递给 de function secuence。

这是我的代码：

spark = SparkSession.builder.appName('sparkdf').getOrCreate()

TyP_dias = spark.createDataFrame([('null','null')], ['MES','NEGOCIO'])

TyP_df0 = TyP_dias.withColumn('FECHA', sf.explode(sf.expr("sequence(to_date('2021-12-01'), to_date('2021-12-31'), interval 1 day)"))).show()

我想要变量中的值 2021-12-01 和 2021-12-31。

类似于：

spark = SparkSession.builder.appName('sparkdf').getOrCreate()

TyP_dias = spark.createDataFrame([('null','null')], ['MES','NEGOCIO'])

eldia1 = '2021-12-01'
eldia2 = '2021-12-31'

TyP_df0 = TyP_dias.withColumn('FECHA', sf.explode(sf.expr("sequence(to_date(eldia1), to_date(eldia2), interval 1 day)"))).show()

得到这个结果：

+----+-------+----------+
| MES|NEGOCIO|     FECHA|
+----+-------+----------+
|null|   null|2021-12-01|
|null|   null|2021-12-02|
|null|   null|2021-12-03|
|null|   null|2021-12-04|
|null|   null|2021-12-05|
|null|   null|2021-12-06|
|null|   null|2021-12-07|
|null|   null|2021-12-08|

但我正在恢复：

cannot resolve 'eldia1' given input columns: [MES, NEGOCIO];

Answer 1

最简单的方法是使用 Python string formatting 将变量内容添加到 sql 表达式中。

TyP_df0 = TyP_dias.withColumn('FECHA', sf.explode(sf.expr(f"sequence(to_date('{eldia1}'), to_date('{eldia2}'), interval 1 day)"))).show()

+----+-------+----------+                                                       
| MES|NEGOCIO|     FECHA|
+----+-------+----------+
|null|   null|2021-12-01|
|null|   null|2021-12-02|
|null|   null|2021-12-03|
|null|   null|2021-12-04|
|null|   null|2021-12-05|
|null|   null|2021-12-06|
|null|   null|2021-12-07|
|null|   null|2021-12-08|
|null|   null|2021-12-09|
|null|   null|2021-12-10|
|null|   null|2021-12-11|
|null|   null|2021-12-12|
|null|   null|2021-12-13|
|null|   null|2021-12-14|
|null|   null|2021-12-15|
|null|   null|2021-12-16|
|null|   null|2021-12-17|
|null|   null|2021-12-18|
|null|   null|2021-12-19|
|null|   null|2021-12-20|
+----+-------+----------+
only showing top 20 rows

如何将变量传递给 Pyspark 序列以生成时间序列？

How to pass a variable into an Pyspark sequence to generate time series?

python

docker

pyspark

apache-zeppelin