只要您使用的是sparkversion2.1或更高版本，您就可以利用这样一个事实：在使用spark2.1时，我们可以使用列值作为参数 pyspark.sql.functions.expr() :
创建一个长度等于的重复逗号伪字符串 diffDays 将此字符串拆分为 ',' 把它变成一个大小的数组 diffDays 使用 pyspark.sql.functions.posexplode() 将此数组与其索引一起分解
最后使用 pyspark.sql.functions.date_add() 将索引值天数添加到 bookingDt 代码：

import pyspark.sql.functions as f

diffDaysDF.withColumn("repeat", f.expr("split(repeat(',', diffDays), ',')"))\
    .select("*", f.posexplode("repeat").alias("txnDt", "val"))\
    .drop("repeat", "val", "diffDays")\
    .withColumn("txnDt", f.expr("date_add(bookingDt, txnDt)"))\
    .show()

# +----------+----------+-------+----------+

# | arrivalDt| bookingDt|vyge_id|     txnDt|

# +----------+----------+-------+----------+

# |2018-01-05|2018-01-01|   1000|2018-01-01|

# |2018-01-05|2018-01-01|   1000|2018-01-02|

# |2018-01-05|2018-01-01|   1000|2018-01-03|

# |2018-01-05|2018-01-01|   1000|2018-01-04|

# |2018-01-05|2018-01-01|   1000|2018-01-05|

# +----------+----------+-------+----------+

好吧，你可以跟着做。
创建仅包含日期的Dataframe： dates_df #从第一天到第二天 bookingDt 最后一个 arrivalDt 然后将这些df与between条件连接起来：

df.join(dates_df, 
  on=col('dates_df.dates').between(col('df.bookindDt'), col('dt.arrivalDt'))
.select('df.*', 'dates_df.dates')

它的工作速度可能比解决方案更快 explode 但是，您需要确定此df的开始和结束日期。10年后，df将只有3650张唱片，不用担心太多。

对于spark 2.4，可以使用sequence创建一个数组，其中包含 bookingDt 以及 arrivalDt . 然后可以分解这个数组。

from pyspark.sql import functions as F

df = df \
  .withColumn('bookingDt', F.col('bookingDt').cast('date')) \
  .withColumn('arrivalDt', F.col('arrivalDt').cast('date'))

df.withColumn('txnDt', F.explode(F.expr('sequence(bookingDt, arrivalDt, interval 1 day)')))\
  .show()

输出：

+-------+----------+----------+----------+
|vyge_id| bookingDt| arrivalDt|     txnDt|
+-------+----------+----------+----------+
|   1000|2018-01-01|2018-01-05|2018-01-01|
|   1000|2018-01-01|2018-01-05|2018-01-02|
|   1000|2018-01-01|2018-01-05|2018-01-03|
|   1000|2018-01-01|2018-01-05|2018-01-04|
|   1000|2018-01-01|2018-01-05|2018-01-05|
+-------+----------+----------+----------+

正如@vvg所建议的：

# I assume, bookindDt has dates range including arrivalDt,

# otherwise you have to find intersection of unique dates of bookindDt and arrivalDt

dates_df = df.select('bookindDt').distinct()
dates_df = dates_df.withColumnRenamed('bookindDt', 'day_of_listing')

listing_days_df = df.join(dates_df, on=dates_df.day_of_listing.between(df.bookindDt, df.arrivalDt))

输出：

+----------+----------+-------+-------------------+
| arrivalDt| bookingDt|vyge_id|day_of_listing     |
+----------+----------+-------+-------------------+
|2018-01-05|2018-01-01|   1000|2018-01-01         |
+----------+----------+-------+-------------------+
|2018-01-05|2018-01-01|   1000|2018-01-02         |
+----------+----------+-------+-------------------+
|2018-01-05|2018-01-01|   1000|2018-01-03         |
+----------+----------+-------+-------------------+
|2018-01-05|2018-01-01|   1000|2018-01-04         |
+----------+----------+-------+-------------------+
|2018-01-05|2018-01-01|   1000|2018-01-05         |
+----------+----------+-------+-------------------+