I have joined a streaming dataframe (consisting of the first four columns below) with a static dataframe (providing the last two columns) to produce a new streaming dataframe (called first_agg_sdf) with the following structure:
+---------+--------+-----+-------------------+-----+------------+
|elementid|metricid|value| epoch|sigma|windowlength|
+---------+--------+-----+-------------------+-----+------------+
| 2| 6|41.01|2018-02-28 16:56:10| 3.3| 5 minutes|
| 2| 6|61.45|2018-02-28 16:56:24| 3.3| 5 minutes|
| 2| 6| 9.13|2018-02-28 16:56:51| 3.3| 5 minutes|
| 2| 6|34.21|2018-02-28 16:57:19| 3.3| 5 minutes|
| 2| 5|43.25|2018-02-28 16:56:10| 3.2| 3 minutes|
| 2| 5| 4.96|2018-02-28 16:56:24| 3.2| 3 minutes|
| 2| 5|22.81|2018-02-28 16:56:51| 3.2| 3 minutes|
| 2| 5| 0.04|2018-02-28 16:57:19| 3.2| 3 minutes|
This has schema:
root
|-- elementid: integer (nullable = true)
|-- metricid: integer (nullable = true)
|-- value: float (nullable = true)
|-- epoch: timestamp (nullable = true)
|-- sigma: double (nullable = true)
|-- windowlength: string (nullable = true)
I then want to produce a sliding window that aggregates over elementid, metricid for a window duration given by the value in the windowlength column for that row.
I produced the following code:
first_agg_window = first_agg_sdf \
.withWatermark("epoch", "30 seconds") \
.groupBy(
window(timeColumn="epoch", windowDuration="windowlength", slideDuration="30 seconds"),
"elementid",
"metricid")
.agg(stddev_pop("value").alias("movingstd"), avg("value").alias("movingavg"), last("value").alias("value"))
The above window aggregation works okay if I provide a string to the windowDuration attribute like so:
windowDuration="5 minutes".
However, if I use the dataframe column value like so:
windowDuration="windowlength"
I get the following error :
Traceback (most recent call last):
File "/home/ec2-user/spark/spark-2.2.0-bin-
hadoop2.7/python/lib/pyspark.zip/pyspark/sql/utils.py", line 63, in deco
File "/home/ec2-user/spark/spark-2.2.0-bin-hadoop2.7/python/lib/py4j-
0.10.4-src.zip/py4j/protocol.py", line 319, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling
z:org.apache.spark.sql.functions.window.
: java.lang.IllegalArgumentException: The provided interval
(windowlength) did not correspond to a valid interval string.at org.apache.spark.sql.catalyst.expressions.TimeWindow.getIntervalInMicroSeconds(TimeWindow.scala:120) at org.apache.spark.sql.catalyst.expressions.TimeWindow$.apply(TimeWindow.scala:148)
at org.apache.spark.sql.functions$.window(functions.scala:2805)
at org.apache.spark.sql.functions$.window(functions.scala:2852)
at org.apache.spark.sql.functions.window(functions.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.lang.reflect.Method.invoke(Unknown Source)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:280)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Unknown Source)
How do I pass in the column value windowlength in each row into attribute windowDuration without generating the above error?
