python - 流式 DataFrames/DataSets 上存在流式聚合时不支持附加输出模式,无水印;;\nJoin Inner
问题描述
我想加入 2 个流,但收到下一个错误,但我不知道如何解决:
流式 DataFrames/DataSets 上存在流式聚合时不支持附加输出模式,无水印;;\nJoin Inner
df_stream = spark.readStream.schema(schema_clicks).option("ignoreChanges", True).option("header", True).format("csv").load("s3://mybucket/*.csv")
display(df_stream.select("SendID", "EventType", "EventDate"))
我想加入 df1 和 df2:
df1 = df_stream \
.withColumn('timestamp', unix_timestamp(col('EventDate'), "MM/dd/yyyy hh:mm:ss aa").cast(TimestampType())) \
.select(col("SendID"), col("timestamp"), col("EventType")) \
.withColumnRenamed("SendID", "SendID_update") \
.withColumnRenamed("timestamp", "timestamp_update") \
.withWatermark("timestamp_update", "1 minutes")
df2 = df_stream \
.withColumn('timestamp', unix_timestamp(col('EventDate'), "MM/dd/yyyy hh:mm:ss aa").cast(TimestampType())) \
.withWatermark("timestamp", "1 minutes") \
.groupBy(col("SendID")) \
.agg(max(col('timestamp')).alias("timestamp")) \
.orderBy('timestamp', ascending=False)
join = df2.alias("A").join(df1.alias("B"), expr(
"A.SendID = B.SendID_update" +
" AND " +
"B.timestamp_update >= A.timestamp " +
" AND " +
"B.timestamp_update <= A.timestamp + interval 1 hour"))
最后,当我以附加模式编写结果时:
join \
.writeStream \
.outputMode("Append") \
.option("checkpointLocation", "s3://checkpointjoin_delta") \
.format("delta") \
.table("test_join")
我收到了上一个错误。
AnalysisException Traceback (最近一次调用最后一次) in () ----> 1 join.writeStream.outputMode("Append").option("checkpointLocation", "s3://checkpointjoin_delta").format("delta")。表(“test_join”)
/databricks/spark/python/pyspark/sql/streaming.py in table(self, tableName) 1137 """ 1138 if isinstance(tableName, basestring): -> 1139 return self._sq(self._jwrite.table(tableName) ) 1140 else: 1141 raise TypeError("tableName can be only a single string")
/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in call (self, *args) 1255 answer = self.gateway_client.send_command(command) 1256 return_value = get_return_value( -> 1257 回答,self.gateway_client,self.target_id,self.name)1258 1259 用于 temp_args 中的 temp_arg:
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw) 67 e.java_exception.getStackTrace()))
解决方案
问题是.groupBy,有必要添加时间戳。例如:
df2 = df_stream \
.withColumn('timestamp', unix_timestamp(col('EventDate'), "MM/dd/yyyy hh:mm:ss aa").cast(TimestampType())) \
.withWatermark("timestamp", "1 minutes") \
.groupBy(col("SendID"), "timestamp") \
.agg(max(col('timestamp')).alias("timestamp")) \
.orderBy('timestamp', ascending=False)
推荐阅读
- c# - 如何在 ASP Core 的静态类中使用“IWebHostEnvironment”
- ada - 为什么 GNAT 会拒绝具有默认判别值的数组类型?
- javascript - 在构建电子时如何提供自己的版本
- mysql - 是否可以使用 . 在名称中使用 ansible 模块?如何?
- c++ - 为什么`std::conj(double d)`不返回`double`而是`std::complex
`? - python - TypeError:“linehistory”对象不是迭代器
- node.js - 可以通过浏览器控制台访问firebase来创建用户吗?
- python - 如何从 AWS 胶水中的 Odata url 读取数据并写入 AWS 中的 S3 存储桶
- r - 在堆叠的 geom_col 图上叠加密度曲线?[包括代码和图像]
- python-3.x - TwitchIO:删除频道中的单个消息