У меня есть код Pyspark ниже. В нем я заполняю пропущенные значения end_date в кадре данных tz_inventory_aud_df2 с датой в далеком будущем. Я получаю минимальное start_date от того же кадра данных. Затем я создаю временной ряд для каждой даты от минимальной даты начала до текущей даты. Я использую udf для создания фрейма данных с этими датами, а затем оставляю соединение этого фрейма данных с tz_inventory_aud_df, чтобы получить сумму полей, отфильтрованных по каждой из дат в фрейме данных, который я создал. Когда я, наконец, пытаюсь записать фрейм данных в виде паркетных файлов, в журнале драйверов появляется ошибка, приведенная ниже. Кто-нибудь знает, в чем причина ошибки, и можете ли вы подсказать, как ее исправить?
код:
tz_inventory_aud_df2=tz_inventory_aud_df.fillna({'end_date':'3018-01-01 00:00:00'})
bs_df=tz_inventory_aud_df.agg({'start_date':'min'})\
.withColumn('min_date',to_date(col('min(start_date)')))
timestamp = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
bs_df = bs_df.withColumn('current_date',to_date(unix_timestamp(lit(timestamp),'yyyy-MM-dd').cast("timestamp")))
# creating time-series dataframe
# UDF
def generate_date_series(start, stop):
return [start + datetime.timedelta(days=x) for x in range(0, (stop-start).days + 1)]
# Register UDF for later usage
spark.udf.register("generate_date_series", generate_date_series, ArrayType(DateType()) )
# mydf is a DataFrame with columns `start` and `stop` of type DateType()
bs_df.createOrReplaceTempView("mydf")
filldate_df=spark.sql("SELECT explode(generate_date_series(min_date, current_date)) as dates FROM mydf")
daily_af_units=filldate_df.alias('a').join(tz_inventory_aud_df2.alias('b'),
(col('b.current_flag')==1)
&(col('a.dates')>=col('b.start_date'))
&(col('a.dates')<col('b.end_date')),
how='inner'
)\
.select(col('b.product_id'),
col('a.dates'),
(col('b.available_units')+col('b.reserved_units')+col('b.packed_and_ready_units')).alias('daily_product_remaining')
)\
.alias('c')\
.groupby(['product_id','dates']).sum()
daily_af_units=daily_af_units.withColumn("daily_product_remaining",daily_af_units["sum(daily_product_remaining)"])
daily_af_units=daily_af_units[['product_id', 'dates', 'daily_product_remaining']]
daily_af_units.write.mode("overwrite").parquet(bckt_pth1+'daily_units_remaining')
ошибка:
2020-03-17 08:03:05,437 WARN [task-result-getter-0] scheduler.TaskSetManager (Logging.scala:logWarning(66)) - Lost task 0.1 in stage 12651.0 (TID 479153, ip-10-100-7-60.glue.dnsmasq, executor 7): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/mnt/yarn/usercache/root/appcache/application_1584428038308_0005/container_1584428038308_0005_01_000013/pyspark.zip/pyspark/worker.py", line 377, in main
process()
File "/mnt/yarn/usercache/root/appcache/application_1584428038308_0005/container_1584428038308_0005_01_000013/pyspark.zip/pyspark/worker.py", line 372, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "/mnt/yarn/usercache/root/appcache/application_1584428038308_0005/container_1584428038308_0005_01_000013/pyspark.zip/pyspark/worker.py", line 248, in <lambda>
func = lambda _, it: map(mapper, it)
File "<string>", line 1, in <lambda>
File "/mnt/yarn/usercache/root/appcache/application_1584428038308_0005/container_1584428038308_0005_01_000013/pyspark.zip/pyspark/worker.py", line 83, in <lambda>
return lambda *a: toInternal(f(*a))
File "/mnt/yarn/usercache/root/appcache/application_1584428038308_0005/container_1584428038308_0005_01_000013/pyspark.zip/pyspark/util.py", line 99, in wrapper
return f(*args, **kwargs)
File "script_2020-03-17-06-55-38.py", line 1839, in generate_date_series
return [start + datetime.timedelta(days=x) for x in range(0, (stop-start).days + 1)]
TypeError: unsupported operand type(s) for -: 'datetime.date' and 'NoneType'
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:452)
at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:81)
Обновление:
tz_inventory_aud_df2=tz_inventory_aud_df[tz_inventory_aud_df['current_flag']==1]\
.fillna({'end_date':'3018-01-01 00:00:00',
'start_date':'1990-01-01 00:00:00'})
bs_df=tz_inventory_aud_df2.agg({'start_date':'min'})\
.withColumn('min_date',to_date(col('min(start_date)')))
timestamp = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
bs_df = bs_df.withColumn('current_date',to_date(unix_timestamp(lit(timestamp),'yyyy-MM-dd').cast("timestamp")))
# creating time-series dataframe
# UDF
def generate_date_series(start, stop):
return [start + datetime.timedelta(days=x) for x in range(0, (stop-start).days + 1)]
# Register UDF for later usage
spark.udf.register("generate_date_series", generate_date_series, ArrayType(DateType()) )
# mydf is a DataFrame with columns `start` and `stop` of type DateType()
bs_df.createOrReplaceTempView("mydf")
filldate_df=spark.sql("SELECT explode(generate_date_series(min_date, current_date)) as dates FROM mydf")
daily_af_units=filldate_df.alias('a').join(tz_inventory_aud_df2.alias('b'),
(col('b.current_flag')==1)
&(col('a.dates')>=col('b.start_date'))
&(col('a.dates')<col('b.end_date')),
how='inner'
)\
.select(col('b.product_id'),
col('a.dates'),
(col('b.available_units')+col('b.reserved_units')+col('b.packed_and_ready_units')).alias('daily_product_remaining')
)\
.alias('c')\
.groupby(['product_id','dates']).sum()
daily_af_units=daily_af_units.withColumn("daily_product_remaining",daily_af_units["sum(daily_product_remaining)"])
daily_af_units=daily_af_units[['product_id', 'dates', 'daily_product_remaining']]