# Write stream to elasticsearch failure **URL:** \ **Category:** Elastic Search\ **Created:** [July 30, 2024, 3:47am UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018 "2024-07-30T03:47:41Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![SaraS](https://avatars.discourse-cdn.com/v4/letter/s/d9b06d/32.png) [@SaraS](https://discuss.elastic.co/u/SaraS)\ **Post date:** [July 30, 2024, 3:47am UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/1 "2024-07-30T03:47:41Z")
Hello. i tried to stream Process pipline with pyspark that read data from kafka topic and write it to Elasticsearch database when i running it in ubuntu environment with python3.10 stream.py ,it is working well and write stream data to elastic but when i try to automate my job with crontab or remote ssh airflow i got below error. Elasticsearch==8.14.0 spark==3.4.1 scala==2.12 `ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)`java.lang.NoSuchMethodError: 'org.apache.spark.sql.catalyst.encoders.ExpressionEncoder org.apache.spark.sql.catalyst.encoders.RowEncoder$.apply(org.apache.spark.sql.types.StructType)' WARNING - at org.elasticsearch.spark.sql.streaming.EsStreamQueryWriter.(EsStreamQueryWriter.scala:50) WARNING - at org.elasticsearch.spark.sql.streaming.EsSparkSqlStreamingSink.$anonfun$addBatch$5(EsSparkSqlStreamingSink.scala:72) WARNING - at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)` ---
**Author:** ![Keith\_Massey](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keith_massey/32/83666_2.png) [@Keith\_Massey](https://discuss.elastic.co/u/Keith_Massey)\ **Post date:** [July 30, 2024, 2:45pm UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/2 "2024-07-30T14:45:06Z")
My guess is that your es-spark jar is incompatible with the version of spark that you are using. Both spark and scala pretty frequently break compatibility with upgrades. What version of es-spark are you using? For the versions you give, you'll want [elasticsearch-spark-30\_2.12-8.14.0.jar](https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch-spark-30_2.12/8.14.0). ---
**Author:** ![Keith\_Massey](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keith_massey/32/83666_2.png) [@Keith\_Massey](https://discuss.elastic.co/u/Keith_Massey)\ **Post date:** [July 30, 2024, 3:28pm UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/3 "2024-07-30T15:28:43Z")
Also, are you sure you're not accidentally using spark 3.5, or picking up spark 3.5 libraries in your classpath somewhere? [Spark 3.5 is not supported yet](https://github.com/elastic/elasticsearch-hadoop/issues/2210). It looks like that method in your stack trace [is present in spark 3.4](https://github.com/apache/spark/blob/branch-3.4/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala#L66). But it is gone from 3.5. ---
**Author:** ![SaraS](https://avatars.discourse-cdn.com/v4/letter/s/d9b06d/32.png) [@SaraS](https://discuss.elastic.co/u/SaraS)\ **Post date:** [July 31, 2024, 5:04am UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/4 "2024-07-31T05:04:31Z")
Thanks for your answer. i use exactly org.elasticsearch\_elasticsearch-spark-30\_2.12-8.14.0.jar as my driver ```auto ______ / __/__ ________ / /__ _\ \/ _ \/ _ `/ __/ '_/ / ___/ .__ /\_,_/_/ /_/\_\ version 3.4.1 /_/ Using Scala version 2.12.17 (OpenJDK 64-Bit Server VM, Java 11.0.23) ``` ````spark .appName("spark") \ .config("spark.sql.legacy.charVarcharAsString", "true") \ .config("spark.jars", "/home/hadoop/W/org.elasticsearch_elasticsearch-spark-30_2.12-8.14.0.jar") \ .config("spark.jars", "/home/hadoop/W/spark-catalyst_2.12-3.4.1.jar") \ .config("spark.jars", "/home/hadoop/W/elasticsearch-hadoop-8.14.0.jar") \ .config("spark.jars", "/home/hadoop/W/spark-sql_2.12-3.4.1.jar") \ .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.1,org.apache.kafka:kafka-clients:3.4.1") \ .getOrCreate()``` ```` ---
**Author:** ![Keith\_Massey](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keith_massey/32/83666_2.png) [@Keith\_Massey](https://discuss.elastic.co/u/Keith_Massey)\ **Post date:** [July 31, 2024, 1:51pm UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/5 "2024-07-31T13:51:31Z")
You don't ordinarily have to pass in the spark sql and catalyst jars, do you? Maybe something is not quite right in the environment? It is the same error we get when we run in spark 3.5 and `org.apache.spark.sql.catalyst.encoders.RowEncoder` is not present. You could try [enabling verbose class loading](https://rangareddy.github.io/SparkVerbose/) to get more information. ---
**Author:** ![SaraS](https://avatars.discourse-cdn.com/v4/letter/s/d9b06d/32.png) [@SaraS](https://discuss.elastic.co/u/SaraS)\ **Post date:** [August 3, 2024, 3:53am UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/6 "2024-08-03T03:53:49Z")
Thank you very much. my problem is solved.The reason was the difference between environment variables between Ubuntu (can see this differences with env command) and the airflow environment . i created a .sh file and added these differences to this bash script . ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [August 31, 2024, 3:54am UTC](https://discuss.elastic.co/t/write-stream-to-elasticsearch-failure/364018/7 "2024-08-31T03:54:43Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.