[
https://issues.apache.org/jira/browse/SPARK-26379?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16748458#comment-16748458
]
Jungtaek Lim commented on SPARK-26379:
--------------------------------------
This looks like also occurred on the master branch. Simpler to reproduce.
{code}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import spark.implicits._
val lines = spark.readStream.format("socket").option("host",
"localhost").option("port", 9999).load()
val final_df = lines.withColumn("ts",lit(current_timestamp()))
val query = final_df.writeStream.format("console").start()
{code}
I'll quickly submit a patch soon.
> Structured Streaming - Exception on adding column to Dataset
> ------------------------------------------------------------
>
> Key: SPARK-26379
> URL: https://issues.apache.org/jira/browse/SPARK-26379
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 2.3.0
> Reporter: Kailash Gupta
> Priority: Major
>
> While using withColumn to add a column to a structured streaming Dataset, I
> am getting following exception:
> org.apache.spark.sql.catalyst.analysis.UnresolvedException: Invalid call to
> dataType on unresolved object, tree: 'timestamp
> Following is sample code
> {code:java}
> final String path = "path_to_input_directory";
> final StructType schema = new StructType(new StructField[] { new
> StructField("word", StringType, false, Metadata.empty()), new
> StructField("count", DataTypes.IntegerType, false, Metadata.empty()) });
> SparkSession sparkSession =
> SparkSession.builder().appName("StructuredStreamingIssue").master("local").getOrCreate();
> Dataset<Row> words = sparkSession.readStream().option("sep",
> ",").schema(schema).csv(path);
> Dataset<Row> wordsWithTimestamp = words.withColumn("timestamp",
> functions.current_timestamp());
> // wordsWithTimestamp.explain(true);
> StreamingQuery query =
> wordsWithTimestamp.writeStream().outputMode("update").option("truncate",
> "false").format("console").trigger(Trigger.ProcessingTime("2
> seconds")).start();
> query.awaitTermination();{code}
> Following are the contents of the file present at _path_
> {code:java}
> a,2
> c,4
> d,2
> r,1
> t,9
> {code}
> This seems working with 2.2.0 release, but not with 2.3.0 and 2.4.0
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
