About techsoln

techsoln · ‎03-06-2020

Hi, I have a spark job in python which writes the data in HDFS. Currently job is running serially where if i have to write 50 files, it will write one by one. I am trying to parallelize it where i want to write all the files in parallel at the same time but it gives an error as below. I am trying to achieve this using python multiprocessing concept where if i will spawn 50 processes and each one will try to write in HDFS, but due to driver memory limit it is failing. Small files gets parallely written but when the file data gets huge it is failing. I understand that it is obviously due to limited driver memory but i am trying to check here of any alternate approach to meet this goal. Error details Container [pid=28341,containerID=container_e130_1583424819028_11130_01_000001] is running beyond physical memory limits. Current usage: 3.4 GB of 2.5 GB physical memory used; 36.1 GB of 5.3 GB virtual memory used. Killing container. Dump of the process-tree for container_e130_1583424819028_11130_01_000001 :

techsoln · ‎05-01-2019

This was failing since my python executable was not in .zip or .egg format. On creation of the executable in .zip format job was accepted.

techsoln · ‎05-01-2019

looks like to run on client mode yoy need to set --master local in your spark2-submit command

techsoln · ‎04-30-2019

From the URL https://www.cloudera.com/documentation/spark2/latest/topics/spark2_known_issues.html#ki_spark_submit_kerberos, only blocking point is that you cannot use principal and keytab configuration with spark2.But definitely there could be more in it. I am also struggling to run pyspark2 using spark2-submit.

techsoln · ‎04-29-2019

Hi , I am upgrading from Spark 1.6.0 to Spark 2.1 on CDH 5.10 platform. I am trying to run spark2-submit command for python implementation and it is failing giving below error. Looks like it is expecting some path property while initilization and creating SparkContext object which is not happening. Error details are as below. Please suggest if any specific configuration is missing or required for spark2. sc = SparkContext(conf=conf) File "/apps/cloudera/parcels/SPARK2-2.1.0.cloudera1-1.cdh5.7.0.p0.120904/lib/spark2/python/lib/pyspark.zip/pyspark/context.py", line 118, in __init__ File "/apps/cloudera/parcels/SPARK2-2.1.0.cloudera1-1.cdh5.7.0.p0.120904/lib/spark2/python/lib/pyspark.zip/pyspark/context.py", line 182, in _do_init File "/apps/cloudera/parcels/SPARK2-2.1.0.cloudera1-1.cdh5.7.0.p0.120904/lib/spark2/python/lib/pyspark.zip/pyspark/context.py", line 249, in _initialize_context File "/apps/cloudera/parcels/SPARK2-2.1.0.cloudera1-1.cdh5.7.0.p0.120904/lib/spark2/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1401, in __call__ File "/apps/cloudera/parcels/SPARK2-2.1.0.cloudera1-1.cdh5.7.0.p0.120904/lib/spark2/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 None.org.apache.spark.api.java.JavaSparkContext. : java.lang.IllegalArgumentException: Can not create a Path from an empty string at org.apache.hadoop.fs.Path.checkPathArg(Path.java:127) at org.apache.hadoop.fs.Path.<init>(Path.java:135) at org.apache.hadoop.fs.Path.<init>(Path.java:94) at org.apache.spark.deploy.yarn.Client.copyFileToRemote(Client.scala:368) at org.apache.spark.deploy.yarn.Client.org$apache$spark$deploy$yarn$Client$$distribute$1(Client.scala:481) at org.apache.spark.deploy.yarn.Client$$anonfun$prepareLocalResources$13.apply(Client.scala:629) at org.apache.spark.deploy.yarn.Client$$anonfun$prepareLocalResources$13.apply(Client.scala:627) at scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:74) at org.apache.spark.deploy.yarn.Client.prepareLocalResources(Client.scala:627) at org.apache.spark.deploy.yarn.Client.createContainerLaunchContext(Client.scala:874) at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:171) at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:56) at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:171) at org.apache.spark.SparkContext.<init>(SparkContext.scala:509) at org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58) at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:526) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:236) at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80) at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69) at py4j.GatewayConnection.run(GatewayConnection.java:214) at java.lang.Thread.run(Thread.java:745)

techsoln · ‎04-29-2019

Thanks, Agreed. I also found the bug details. Based on the URL https://spark.apache.org/docs/1.6.0/#downloading you shared, it contains details which says it is compatible with 2.6+ and 3.1+ which is totally misleading since 3.6 is 3.1+ I have started working to upgrade my app to spark 2. Any suggestiosn on Spark 1.6 to Spark 2 migration guide on Cloudera cluster

techsoln · ‎04-29-2019

Hi All, We are currently using Spark 1.6 on CDH 5.10 platform. We are currently upgrading from python 2.7 to python 3.6 using anaconda distribution. While i try to do spark-submit in client mode the process is failing giving below error - File "/apps/cloudera/parcels/CDH-5.10.1-1.cdh5.10.1.p0.10/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 381, in namedtuple TypeError: namedtuple() missing 3 required keyword-only arguments: 'verbose', 'rename', and 'module' We are not very clear about the cause of the failure. We have checked Spark documentation and it says that Spark 1.6.0 is compatible with python 3.0+. Any thoughts or suggestions on this would be helpful ? Thanks Hemil

techsoln · ‎03-12-2019

Hi, I am running spark-submit job on yarn cluster during which it is uploading dependent jars in default HDFS staging directory which is /user/<user id>/.sparkStaging/<yarn applicationId>/*.jar. On verification during spark-submit job, i see that jar is getting uploaded but spark-submit is failing with below error - file owner and group belongs to the same id using which spark-submit is performed. I also tried using configuration parameter spark.yarn.StagingDir but even that didn't helped. Your professional inputs will help in addressing this issue. Error stack trace - ========================= Diagnostics: File does not exist: hdfs://user/<user id>/.sparkStaging/<yarn application_id>/chill-java-0.5.0.jar java.io.FileNotFoundException: File does not exist: hdfs://user/<user id>/.sparkStaging/<yarn application_id>/chill-java-0.5.0.jar at org.apache.hadoop.hdfs.DistributedFileSystem$20.doCall(DistributedFileSystem.java:1257) at org.apache.hadoop.hdfs.DistributedFileSystem$20.doCall(DistributedFileSystem.java:1249) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1249) at org.apache.hadoop.yarn.util.FSDownload.copy(FSDownload.java:251) at org.apache.hadoop.yarn.util.FSDownload.access$000(FSDownload.java:61) at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:359) at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:357) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1920) at org.apache.hadoop.yarn.util.FSDownload.call(FSDownload.java:356) at org.apache.hadoop.yarn.util.FSDownload.call(FSDownload.java:60) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Thanks Hemil

techsoln · ‎07-12-2017

Hi, I have a file which keeps getting appended with new lines. Its kind of log file with data in it. I need to read the data from the file store it in database. My question is, how will i read only unprocessed records from the file and do not read already processed records using Apache Spark framework. Thanks

techsoln · ‎06-12-2017

Thanks so much for your inputs. So is there any recommended solution (software based) to achieve rack failure fault tolerance capability with 2 racks. !! 🙂

techsoln · ‎06-09-2017

Hi, Thanks for your response. I understand zookeeper requirement about having the majority of known ZK nodes to be up. With the example you mentioned, it is fine when i have total 3 nodes and 1 node goes down, since 2F +1 = 3 and it is = our total nodes in quorum we had setup or you can also say that we still have more active nodes then failed nodes. But when we have inter rack communication(specially 2 rack), it is different. As i said, Let say - I have active 3 ZK nodes on Rack A and 4 on rack B. In this case my quorum size is 7 and if my Rack B goes down, i am left with only 3 ZK nodes active. In this case 2F+1 says that, i should have setup 9 nodes in my cluster (F = 4) whereas i have setup 7 nodes. So zookeeper cluster across 2 Racks is not going to satisfy above rule since no of failed nodes are more then active nodes. Is there anyways concept of Kafka mirroring or setting up isolated zookeeper cluster and then cross connect zookeeper cluster so that when rack B goes down, rack A zookeeper takes over the task help in this 2 rack communication ?

techsoln · ‎06-09-2017

I was referring to the concept of Kafka Mirroring - https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330 If this is implemented then can it be used to achieve rack fault tolerance across 2 racks cluster setup ? My limit is to implement scalable & fault tolerant messaging layer with 2 Rack infrastructure.

techsoln · ‎06-08-2017

Hi, I understand that Kafka mirroring is replicating data across different data centers. I have some questions here as below - 1) Does these different data centers needs different zookeeper ensemble ? 2) Can these Kafka mirroring concept be a good option if we want to replicate data across 2 racks which are part of data center ? Thanks

techsoln · ‎06-08-2017

Hi, Thanks again. This gives me an idea of setting up Kafka cluster in VM environment which is then connected to my actual existing servers. How about Kafka mirroring ? Can these be useful concept of achieving fault tolerance in case of rack failure ? And how will zookeeper ensemble looks in it ? Will it be standalone on both the rack and the target rack zookeeper gets activated when source rack goes down ?

techsoln · ‎06-08-2017

Hi, Thanks for your response. So can i say that Kafka which is managed zookeeper. is not a right messaging platform for 2 Rack network infrastructure to achieve seamless rack fault tolerant mechanism. Or any other mechanism exists ?

techsoln · ‎06-08-2017

Hi, I am trying to understand the impact and design for zookeeper setup since Kafka is dependent on zookeeper for its operations. Zookeeper specifies 2F+1 no of nodes to be setup for reliable fault tolerance. Consider that If I have 2 racks and I setup 4 nodes on rack A and 5 on rack B (Total 9 zookeeper nodes) and rack B goes down (5 zookeeper nodes goes down). In that case with the requirement of 2F+1, it needs 11 zookeeper nodes where as I have only 9 nodes. So zookeeper in case of rack failure with higher no of nodes will not be able to sustain which will impact Kafka cluster behavior. Can you please provide your inputs on how to better setup zookeeper so that Kafka can work seamlessly in case of 2 rack infrastructure

techsoln · ‎06-08-2017

Hi, I am trying to understand the impact and design for zookeeper setup since Kafka is dependent on zookeeper for its operations. Zookeeper specifies 2F+1 no of nodes to be setup for reliable fault tolerance. Consider that If I have 2 racks and I setup 4 nodes on rack A and 5 on rack B (Total 9 zookeeper nodes) and rack B goes down (5 zookeeper nodes goes down). In that case with the requirement of 2F+1, it needs 11 zookeeper nodes where as I have only 9 nodes. So zookeeper in case of rack failure with higher no of nodes will not be able to sustain which will impact Kafka cluster behavior. Can you please provide your inputs on how to better setup zookeeper so that Kafka can work seamlessly in case of 2 rack infrastructure.

techsoln · ‎06-08-2017

Hi, We have a 2 rack data center and both the rack has multiple servers. We are looking to setup the zookeeper nodes on these servers and it should provide failover mechanism in case of any rack failure. Based on zookeeper admin requirement, it needs 2F+1 number of nodes and odd number of nodes in a quorum. So in my case let say, i setup 3 zookeeper nodes on rack A and 4 on rack B. So total nodes in quorum is 7. Based on 2F+1 if rack B goes down then F is 4 here and it needs 9 nodes. If Rack A goes down then it 7. So the scenario of complete rack failure(Rack B) with with larger number of nodes will not sustain. I want to understand if zookeeper is preferable solution to be used in case of applications with 2 rack datacenter. If yes, how can i build the infrastructure in such case.

techsoln · ‎06-08-2017

Hi, We have a 2 rack data center and both the rack has multiple servers. We are looking to setup the zookeeper nodes on these servers and it should provide failover mechanism in case of any rack failure. Based on zookeeper admin requirement, it needs 2F+1 number of nodes and odd number of nodes in a quorum. So in my case let say, i setup 3 zookeeper nodes on rack A and 4 on rack B. So total nodes in quorum is 7. Based on 2F+1 if rack B goes down then F is 4 here and it needs 9 nodes. If Rack A goes down then it 7. So the scenario of complete rack failure(Rack B) with with larger number of nodes will not sustain. I want to understand if zookeeper is preferable solution to be used in case of applications with 2 rack datacenter. If yes, how can i build the infrastructure in such case. Please reach me out if any information is needed. Thanks Hemil

Online	Offline
Last Visited	‎03-24-2020 02:48 PM

Date Registered	‎06-08-2017 08:33 AM
Last Visited	‎03-24-2020 02:48 PM
Total Messages Posted	19

Re: spark2-submit using pyspark fails

Python Spark job gives error -running beyond memor...

Re: spark2-submit using pyspark fails

Re: Is it impossible to run a Spark2 job with --ma...

Re: Is it impossible to run a Spark2 job with --ma...

spark2-submit using pyspark fails

Re: pyspark fails with python 3.6

pyspark fails with python 3.6

spark-submit fails giving FileNotFoundException

Apache Spark - Filtering Unprocessed File Data

Re: Zookeeper Setup for 2 Rack Data center

Re: Zookeeper Setup for 2 Rack Data Center

Re: Zookeeper Setup for 2 Rack Data center

Does Kafka Mirroring needs different zookeeper ens...

Re: Zookeeper Setup for 2 Rack Data center

Re: Zookeeper Setup for 2 Rack Data center

Re: Can anyone explain Kafka rack awareness featur...

Re: Can anyone explain Kafka rack awareness featur...

Zookeeper Setup for 2 Rack Data Center

Zookeeper Setup for 2 Rack Data center