Spark 3 connect to Hive 1.2
Greetings, Our standalone Spark 3 cluster is trying to connect to Hadoop 2.6 cluster running Hive server 1.2 (/usr/hdp/2.6.2.0-205/hive/lib/hive-service-1.2.1000.2.6.2.0-205.jar) import org.apache.spark.sql.functions._ import java.sql.Timestamp val df1 = spark.createDataFrame( Seq( ("id1", "v2", "notshared", Timestamp.valueOf("2019-09-13 10:00:00"), false, 1, "2019-09-13"), ("id2", "v3", "notshared", Timestamp.valueOf("2019-09-13 09:00:00"), false, 2, "2019-09-13"), ("id2", "v4", "notshared", Timestamp.valueOf("2019-09-14 11:00:00"), false, 3, "2019-09-14"), ("id2", "v5", "notshared", Timestamp.valueOf("2019-09-14 13:00:00"), false, 4, "2019-09-14"), ("id3", "v4", "notshared", Timestamp.valueOf("2019-09-14 17:00:00"), false, 5, "2019-09-14"), ("id4", "v1", "notshared", Timestamp.valueOf("2019-09-15 19:00:00"), false, 6, "2019-09-15"))).toDF("user_id", "col2", "pidd", "land_ts", "deleted","offset", "partition") df1.write.mode("overwrite").saveAsTable("db.spark3_test") when running above code, is throws the error : org.apache.spark.sql.AnalysisException: > org.apache.hadoop.hive.ql.metadata.HiveException: Unable to fetch table > spark_27686. Invalid method name: 'get_table_req'; > at > org.apache.spark.sql.hive.HiveExternalCatalog.withClient(HiveExternalCatalog.scala:110) > at I assume this is caused because Spark 3 ships "hive-metastore-2.3.7.jar". To work with Hive Server 1.2 can I use "hive-metastore-1.2.1.spark2.jar" from Spark 2.4 distribution ? Do I need any other dependencies ?
How to optimize the configuration and/or code to solve the cache overloading issue?
I met a trouble in using spark structured streaming. The usercache is continuously consumed due to the join operation without releasing. How can I optimize the configuration and/or code to solve this problem? Spark Cluster in AWS EMR. 1 master node, m4.xlarge, 4 core, 16GB 2 core nodes, m4.xlarge, 4 core, 16GB yarn configuration: 'yarn.nodemanager.disk-health-checker.enable':'true', 'yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage':'95.0', 'yarn.nodemanager.localizer.cache.cleanup.interval-ms': '10', 'yarn.nodemanager.localizer.cache.target-size-mb': '1024', 'yarn.nodemanager.pmem-check-enabled': 'false', 'yarn.nodemanager.vmem-check-enabled': 'false', 'yarn.log-aggregation.retain-seconds': '12000' spark-submit --deploy-mode cluster --num-executors 3 --executor-memory 8G --executor-cores 2 Code snippet: //Disable broadcast join spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1) val monitoring_stream = volume_df_filtered.writeStream .trigger(Trigger.ProcessingTime("120 seconds")) .foreachBatch { (batchDF: DataFrame, batchId: Long) => if(!batchDF.isEmpty) { monitoring_df = monitoring_df.join(batchDF, monitoring_df("id") === batchDF("id"), "left").select(monitoring_df("id"), monitoring_df("total_volume"), batchDF("volume")).na.fill(0) //This join operation consumes the usercache continusly. monitoring_df = monitoring_df.withColumn("total_volume", monitoring_df("total_volume")+monitoring_df("volume")) monitoring_df = monitoring_df.repartition(6) batchDF.unpersist() spark.catalog.clearCache() }
Spark DataFrame Creation
Sorry if this is the wrong place for this. I am trying to debug an issue with this library: https://github.com/springml/spark-sftp When I attempt to create a dataframe: spark.read. format("com.springml.spark.sftp"). option("host", "..."). option("username", "..."). option("password", "..."). option("fileType", "csv"). option("inferSchema", "true"). option("tempLocation","/srv/spark/tmp"). option("hdfsTempLocation","/srv/spark/tmp"); .load("...") What I am seeing is that the download is occurring on the spark driver not the spark worker, This leads to a failure when spark tries to create the DataFrame on the worker. I'm confused by the behavior. my understanding was that load() was lazily executed on the Spark worker. Why would some elements be executing on the driver? Thanks for your help -- Mark Bidewell http://www.linkedin.com/in/markbidewell
Re: Spark DataFrame Creation
You'd probably do best to ask that project, but scanning the source code, that looks like it's how it's meant to work. It downloads to a temp file on the driver then copies to distributed storage then returns a DataFrame for that. I can't see how it would be implemented directly over sftp as there would be so many pieces missing - locality, blocking, etc. On Wed, Jul 22, 2020 at 4:48 PM Mark Bidewell wrote: > > Sorry if this is the wrong place for this. I am trying to debug an issue > with this library: > https://github.com/springml/spark-sftp > > When I attempt to create a dataframe: > > spark.read. > format("com.springml.spark.sftp"). > option("host", "..."). > option("username", "..."). > option("password", "..."). > option("fileType", "csv"). > option("inferSchema", "true"). > option("tempLocation","/srv/spark/tmp"). > option("hdfsTempLocation","/srv/spark/tmp"); > .load("...") > > What I am seeing is that the download is occurring on the spark driver not > the spark worker, This leads to a failure when spark tries to create the > DataFrame on the worker. > > I'm confused by the behavior. my understanding was that load() was lazily > executed on the Spark worker. Why would some elements be executing on the > driver? > > Thanks for your help > -- > Mark Bidewell > http://www.linkedin.com/in/markbidewell - To unsubscribe e-mail: user-unsubscr...@spark.apache.org
Re: Spark DataFrame Creation
Hi Mark, On Wed, Jul 22, 2020 at 4:49 PM Mark Bidewell wrote: > > Sorry if this is the wrong place for this. I am trying to debug an issue > with this library: > https://github.com/springml/spark-sftp > > When I attempt to create a dataframe: > > spark.read. > format("com.springml.spark.sftp"). > option("host", "..."). > option("username", "..."). > option("password", "..."). > option("fileType", "csv"). > option("inferSchema", "true"). > option("tempLocation","/srv/spark/tmp"). > option("hdfsTempLocation","/srv/spark/tmp"); > .load("...") > > What I am seeing is that the download is occurring on the spark driver not > the spark worker, This leads to a failure when spark tries to create the > DataFrame on the worker. > > I'm confused by the behavior. my understanding was that load() was lazily > executed on the Spark worker. Why would some elements be executing on the > driver? Looking at the code, it appears that your sftp plugin downloads the file to a local location and opens from there. https://github.com/springml/spark-sftp/blob/090917547001574afa93cddaf2a022151a3f4260/src/main/scala/com/springml/spark/sftp/DefaultSource.scala#L38 You may have more luck with an sftp hadoop filesystem plugin that can read sftp:// URLs directly. Cheers Andrew > > Thanks for your help > -- > Mark Bidewell > http://www.linkedin.com/in/markbidewell - To unsubscribe e-mail: user-unsubscr...@spark.apache.org
Spark Job Fails with Unknown Error writing to S3 from AWS EMR
Hi, Can someone help to resolve this issue? Thank you in advance. Error logs : java.io.EOFException: Unexpected EOF while trying to read response from server at org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) at org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) at org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) 20/07/22 22:43:37 WARN DataStreamer: Error Recovery for BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1498 in pipeline [DatanodeInfoWithStorage[172.19.222.182:50010,DS-7783002b-d57a-43a3-9d91-9934e2d063f8,DISK], DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK], DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK]]: datanode 0(DatanodeInfoWithStorage[172.19.222.182:50010,DS-7783002b-d57a-43a3-9d91-9934e2d063f8,DISK]) is bad. 20/07/22 22:44:58 WARN DataStreamer: Exception for BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1499 java.io.EOFException: Unexpected EOF while trying to read response from server at org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) at org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) at org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) 20/07/22 22:44:58 WARN DataStreamer: Error Recovery for BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1499 in pipeline [DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK], DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK], DatanodeInfoWithStorage[172.19.222.180:50010,DS-40b1c81b-18d1-4d8d-ab49-11904f3dd23c,DISK]]: datanode 0(DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK]) is bad. 20/07/22 22:47:00 WARN DataStreamer: Exception for BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1500 java.io.EOFException: Unexpected EOF while trying to read response from server at org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) at org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) at org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) 20/07/22 22:47:00 WARN DataStreamer: Error Recovery for BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1500 in pipeline [DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK], DatanodeInfoWithStorage[172.19.222.180:50010,DS-40b1c81b-18d1-4d8d-ab49-11904f3dd23c,DISK], DatanodeInfoWithStorage[172.19.223.55:50010,DS-3e5dc677-cd1d-49fc-b50a-4b058ae298aa,DISK]]: datanode 0(DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK]) is bad. 20/07/22 22:47:03 INFO MemoryStore: Block broadcast_4 stored as values in memory (estimated size 18.0 GB, free 24.5 GB) -- Thanks, Koti Reddy Nusum, +1-(660) 541-5623.
Re: Spark Job Fails with Unknown Error writing to S3 from AWS EMR
We faced this similar situation with jre 8u262 try reverting back... On Thu, Jul 23, 2020, 5:18 AM koti reddy wrote: > Hi, > > Can someone help to resolve this issue? > Thank you in advance. > > Error logs : > > java.io.EOFException: Unexpected EOF while trying to read response from server > at > org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) > at > org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) > at > org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) > 20/07/22 22:43:37 WARN DataStreamer: Error Recovery for > BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1498 in pipeline > [DatanodeInfoWithStorage[172.19.222.182:50010,DS-7783002b-d57a-43a3-9d91-9934e2d063f8,DISK], > > DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK], > > DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK]]: > datanode > 0(DatanodeInfoWithStorage[172.19.222.182:50010,DS-7783002b-d57a-43a3-9d91-9934e2d063f8,DISK]) > is bad. > 20/07/22 22:44:58 WARN DataStreamer: Exception for > BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1499 > java.io.EOFException: Unexpected EOF while trying to read response from server > at > org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) > at > org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) > at > org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) > 20/07/22 22:44:58 WARN DataStreamer: Error Recovery for > BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1499 in pipeline > [DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK], > > DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK], > > DatanodeInfoWithStorage[172.19.222.180:50010,DS-40b1c81b-18d1-4d8d-ab49-11904f3dd23c,DISK]]: > datanode > 0(DatanodeInfoWithStorage[172.19.223.27:50010,DS-1bc8cea7-9c28-4869-aada-55b7d0b0680c,DISK]) > is bad. > 20/07/22 22:47:00 WARN DataStreamer: Exception for > BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1500 > java.io.EOFException: Unexpected EOF while trying to read response from server > at > org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:402) > at > org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:213) > at > org.apache.hadoop.hdfs.DataStreamer$ResponseProcessor.run(DataStreamer.java:1073) > 20/07/22 22:47:00 WARN DataStreamer: Error Recovery for > BP-439833631-172.19.222.143-1595381416559:blk_1073742309_1500 in pipeline > [DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK], > > DatanodeInfoWithStorage[172.19.222.180:50010,DS-40b1c81b-18d1-4d8d-ab49-11904f3dd23c,DISK], > > DatanodeInfoWithStorage[172.19.223.55:50010,DS-3e5dc677-cd1d-49fc-b50a-4b058ae298aa,DISK]]: > datanode > 0(DatanodeInfoWithStorage[172.19.223.199:50010,DS-880d5121-16a2-465e-ad20-ca99f4287770,DISK]) > is bad. > 20/07/22 22:47:03 INFO MemoryStore: Block broadcast_4 stored as values in > memory (estimated size 18.0 GB, free 24.5 GB) > > -- > Thanks, > Koti Reddy Nusum, > +1-(660) 541-5623. > >