Hello - i'm using PyCharm 2019, python 2.7.6, Apache Spark 2.2.0 for pyspark development, and running into issues when i try to run any spark code. Code : #! /bin/python import os import sys from pyspark.sql import SparkSession from pyspark import SparkConf, SparkContext import pandas as pd import numpy as np import pyarrow.parquet as pq import pyarrow as pa import pprint as pp from pyspark.sql.functions import col, lit import datetime today = datetime.date.today().__format__("yyyymmdd") print (" formatted date => " + today) # d = {'a':1, 'b':2, 'c':3} d = [(1, "20190101", "99990101", "false"), (2, "20190501", "99990101", "false"), (3, "20190101", "20190414", "false"), (3, "20190415", "99990101", "false")] print(os.environ) print(" spark_home -> ", os.environ['SPARK_HOME']) print(" pythonpath -> ", os.environ['PYTHONPATH']) print(" PYSPARK_SUBMIT_ARGS -> ", os.environ['PYSPARK_SUBMIT_ARGS']) os.environ['PYSPARK_SUBMIT_ARGS'] = "--master spark://Karans-MacBook-Pro-4.local:7077 pyspark-shell" print(" JAVA_HOME => " + os.environ['JAVA_HOME']) print(" PATH => " + os.environ['PATH']) print(" CLASSPATH => " + os.environ['CLASSPATH']) sys.path.append(os.path.join(os.environ['SPARK_HOME'], "python")) sys.path.append(os.path.join(os.environ['SPARK_HOME'], "python/lib/py4j-0.10.4-src.zip")) # ERROR OBTAINED WHEN I CREATE SaprkSession object spark = SparkSession.builder.master("local").appName("CreatingDF").getOrCreate() sparkdf = spark.createDataFrame(d, ['pnalt', 'begda', 'endda', 'ref_flag']) print(sparkdf) ERROR (when i create the SparkSession object) 19/05/15 16:58:06 ERROR SparkUncaughtExceptionHandler: Uncaught exception in thread Thread[main,5,main] java.util.NoSuchElementException: key not found: _PYSPARK_DRIVER_CALLBACK_HOST     at scala.collection.MapLike$class.default(MapLike.scala:228)     at scala.collection.AbstractMap.default(Map.scala:59)     at scala.collection.MapLike$class.apply(MapLike.scala:141)     at scala.collection.AbstractMap.apply(Map.scala:59)     at org.apache.spark.api.python.PythonGatewayServer$$anonfun$main$1.apply$mcV$sp(PythonGatewayServer.scala:50)     at org.apache.spark.util.Utils$.tryOrExit(Utils.scala:1262)     at org.apache.spark.api.python.PythonGatewayServer$.main(PythonGatewayServer.scala:37)     at org.apache.spark.api.python.PythonGatewayServer.main(PythonGatewayServer.scala)     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)     at java.lang.reflect.Method.invoke(Method.java:497)     at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:755)     at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180)     at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205)     at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)     at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) Traceback (most recent call last):   File "/Users/karanalang/PycharmProjects/Python2/karan/python2/falcon/py_createDF.py", line 55, in <module>     spark = SparkSession.builder.master("local").appName("CreatingDF").getOrCreate()   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/sql/session.py", line 173, in getOrCreate     sc = SparkContext.getOrCreate(sparkConf)   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/context.py", line 367, in getOrCreate     SparkContext(conf=conf or SparkConf())   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/context.py", line 133, in __init__     SparkContext._ensure_initialized(self, gateway=gateway, conf=conf)   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/context.py", line 316, in _ensure_initialized     SparkContext._gateway = gateway or launch_gateway(conf)   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/java_gateway.py", line 46, in launch_gateway     return _launch_gateway(conf)   File "/Users/karanalang/PycharmProjects/Python2/venv/lib/python2.7/site-packages/pyspark/java_gateway.py", line 108, in _launch_gateway     raise Exception("Java gateway process exited before sending its port number") Exception: Java gateway process exited before sending its port number forums on internet seem to be suggesting that it is a version related issue, however, i'm not sure how to debug/fix this issue. Any help on this is really appreciated. Attaching screenshot of the Pycharm Project interpretor as well. ... View more

Hello, Here is the scenario : I've 2 branches created - Branch-b1, Branch-b2 for 2 commits Both the branches have Different files changed & committed .. when i commit files into b1, the 2 files specific to b1 get committed Checking in files to Branch-b1 git checkout -b "Branch-b1" git add /path/file1.scala git add /path/file2.scala git commit -m "<Comment>" git push -u origin "Branch-b1" However, when i create b2 & commit the 6 files specific to b2, 6 + 5 files are getting committed Checking in files to Branch-b2 git checkout -b "Branch-b2" git add /path/file3.scala git add /path/file4.scala git commit -m "<Comment>" git push -u origin "Branch-b2" In the second case, what is getting checked in is - files fro Branch-b1 + files from Branch-b2 In fact, when i create a new branch (eg b3) and donot commit any files, & i do a diff between the branch b3 & master . i see the files checked in for branch b1 eg. git checkout -b "Branch-b3" git diff Branch-b3 master --- shows the files from Branch-b1 /path/file1.scala /path/file2.scala Any ideas on what i'm doing wrong ? and how to fix this ? ... View more

hello - i'm trying to read an avro schema stored in a location .. here is the code i tried - val html = Source.fromURL("https://drive.google.com/open?id=1evVW9bTy5bOqq6h_oCyhXDFv9Fuz6zUn") val s = html.mkString println("schema is => " + s) This seems to be printing contents shown below, instead of reading the schema stored in the file. How do i read just the schema ? ... View more

I have an Avro generated class - com.avro.Person which has a method -> getClassSchema. I’m passing className to a method, and in the method I need to get the Avro schema. Here is the code I’m using: val pr = Class.forName(productCls) // where productCls = classOf[Product].getName How do I call the method getClassSchema of this class ? Additional Info : Links to Person.java & Person.avsc Person.java [https://drive.google.com/open?id=1LOa_muN0tFSEm-VReOBDh3ozU3ufmLZQ] Person.avsc [https://drive.google.com/open?id=1evVW9bTy5bOqq6h_oCyhXDFv9Fuz6zUn][1] In Person.java, there is a method public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; } I'm trying to call this method to get the AVRO schema corresponding to class - Person Code import com.avro.Person val m1 = Map("schema.class" -> classOf[Person].getName) This Map is passed onto a function .. say foo, and i'm trying to get the schema based on the class def foo(m : Map[String, String]){ val personClass = Map("schema.class") val pr = Class.forName(personClass) //TBD - to get the schema of the avro class } Now using this "pr", i need to get the avro schema of the Person avro, hence i'm trying to call method getClassSchema, which returns the schema How do i do this ? any help on this is really appreciated. ... View more

Hello All - I'm getting this error, when publishing messages to Kafka topic using SSL mode, Command to publish messages : /usr/hdp/2.5.3.0-37/kafka/bin/kafka-producer-perf-test.sh --messages 1000000 --message-size 1000 --topics mmtest4 \ --broker-list <host1>:9093,<host2>:9093,<host3>:9093, \ --threads 1 --compression-codec 3 --batch-size 10000 \ --security-protocol SSL --show-detailed-stats Error message :[2017-12-21 19:48:49,846] WARN Fetching topic metadata with correlation id 11 for topics [Set(mmtest4)] from broker [BrokerEndPoint(0,<host1>,9093)] failed (kafka.client.ClientUtils$) java.io.EOFException at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:99) at kafka.network.BlockingChannel.readCompletely(BlockingChannel.scala:140) at kafka.network.BlockingChannel.receive(BlockingChannel.scala:131) at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:84) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$doSend(SyncProducer.scala:81) at kafka.producer.SyncProducer.send(SyncProducer.scala:126) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59) at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:83) at kafka.producer.BrokerPartitionInfo.getBrokerPartitionInfo(BrokerPartitionInfo.scala:50) at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$getPartitionListForTopic(DefaultEventHandler.scala:206) at kafka.producer.async.DefaultEventHandler$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:170) at kafka.producer.async.DefaultEventHandler$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:169) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) at kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:169) at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:101) at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:106) at kafka.producer.async.ProducerSendThread$anonfun$processEvents$3.apply(ProducerSendThread.scala:89) at kafka.producer.async.ProducerSendThread$anonfun$processEvents$3.apply(ProducerSendThread.scala:69) at scala.collection.Iterator$class.foreach(Iterator.scala:727) at scala.collection.AbstractIterator.foreach(Iterator.scala:1157) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:68) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:46) Seems SSL is setup correctly, when i run the following command, i get expected result -> <strong><em>openssl s_client -debug -connect host1:9093 -tls1</em></strong> Setting done in the server.properties (using Ambari) -> listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093 advertised.listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093 ssl.truststore.location=/etc/kafka/truststore/kafka.server.truststore.jks ssl.truststore.password=password ssl.keystore.location=/etc/kafka/truststore//kafka.server.keystore.jks ssl.keystore.password=password ssl.key.password=password These setting seem to be working on another environment, however - on this env (Prod), it seem to be giving the error shown above. Any ideas on what need to be done to debug/fix the error ? ... View more

hello - i'm getting error in querying Hive table (data in Parquet format) : select * from powerpoll_k1 where year=2017 and month=12 and day=11 limit 5; Error: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 102.0 failed 4 times, most recent failure: Lost task 0.3 in stage 102.0 (TID 20602, msc02-jag-dn-016.uat.gdcs.apple.com): java.lang.UnsupportedOperationException: org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainIntegerDictionaryat org.apache.parquet.column.Dictionary.decodeToLong(Dictionary.java:52)at org.apache.spark.sql.execution.vectorized.OnHeapColumnVector.getLong(OnHeapColumnVector.java:274)at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370)at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:246)at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:240)at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)at org.apache.spark.scheduler.Task.run(Task.scala:86)at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)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) Table DDL is shown below : CREATE EXTERNAL TABLE `powerpoll_k1`(`topic_k` varchar(255), `partition_k` int, `offset_k` bigint, `timestamp_k` timestamp, `deviceid` bigint, `devicename` varchar(50), `deviceip` varchar(128), `peerid` int, `objectid` int, `objectname` varchar(256), `objectdesc` varchar(256), `oid` varchar(50), `pduoutlet` varchar(50), `pluginid` int, `pluginname` varchar(255), `indicatorid` int, `indicatorname` varchar(255), `format` int, `snmppollvalue` varchar(128) COMMENT 'value in kafka avsc', `time` double, `clustername` varchar(50) COMMENT 'rpp or power', `peerip` varchar(50))COMMENT 'external table at /apps/hive/warehouse/amp.db/sevone_power'PARTITIONED BY (`year` int, `month` int, `day` int) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'WITH SERDEPROPERTIES ('serialization.format' = '1')STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'LOCATION 'hdfs://application/apps/hive/warehouse/amp.db/power'TBLPROPERTIES ('transient_lastDdlTime' = '1513022286') any ideas on what the issue is ? ... View more

Hello All - I was able to set up SSL for the Kafka brokers, using OpenSSL. however, I'm having issues with setting up SSL using the pem file (i.e. SSL certificate - certified by CA, provided by the company) Here is what i've done - created the server/client keystore & truststore files and imported the provided cert.pem file keytool -keystore kafka.server.keystore.jks -alias localhost -validity 365 -genkey keytool -keystore kafka.server.truststore.jks -alias CARoot -import -file cert.pem keytool -keystore kafka.client.truststore.jks -alias CARoot -import -file cert.pem keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file cert.pem keytool -keystore kafka.client.keystore.jks -alias localhost -validity 365 -genkey keytool -keystore kafka.client.keystore.jks -alias CARoot -import -file cert.pem I've a console producer pushing data in to the topic, and gives error as shown below -> Any ideas on what the issue might be ? Caused by: javax.net.ssl.SSLProtocolException: Handshake message sequence violation, state = 1, type = 1 at sun.security.ssl.ServerHandshaker.processMessage(ServerHandshaker.java:213) at sun.security.ssl.Handshaker.processLoop(Handshaker.java:1026) at sun.security.ssl.Handshaker$1.run(Handshaker.java:966) at sun.security.ssl.Handshaker$1.run(Handshaker.java:963) at java.security.AccessController.doPrivileged(Native Method) at sun.security.ssl.Handshaker$DelegatedTask.run(Handshaker.java:1416) at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:336) at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:417) ... 7 more thanks for the help in advance! ... View more

Hello All - i've a basic word count Kafka streams code, which reads data from the input topic, splits the data (per the separator) and outputs the data into the output topic. I've a console producer, which is putting data into the input topic (kstreams3), and a console consumer which is reading the (split) data from output topic(kstreams4).. The code seems to be working fine, but on the console consumer, i'm Not able to see the text. It seems to be printing blanks, instead of the actual text .. (though, console consumer seems to be consuming the correct number of lines) Any pointers on what the issue is ? Code -> Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streamswordcount-application"); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> textLines = builder.stream("kstreams3"); KTable<String, Long> wordCounts = textLines .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+"))) .groupBy((key, word)-> word) .count("Counts"); System.out.println(" Kstreams - wordCount " + wordCounts); wordCounts.to(Serdes.String(),Serdes.Long(),"kstreams4"); KafkaStreams streams = new KafkaStreams(builder, props); streams.start(); ... View more

Hi All - I've Kafka 0.10 .. I've enabled SSL(Non-kerberized) for Kafka Broker on Node 4, and i'm able to produce/consume messages using console-producer & console-consumer from Node 4. However, when i'm having issues enabling ssl connection between Node 4 & Node 5 & try to consume messages from Node5 (using console-consumer), i'm facing issues. here are the steps - On node4 : Started console-producer, publishing on SSL port (9192) $CONFLUENT/bin/kafka-console-producer --broker-list node4:9192 --topic ssl --producer.config client-ssl.properties Started console-consumer on node4, consuming messages from console producer (port 9192) $CONFLUENT/bin/kafka-console-consumer --bootstrap-server node4:9192 --topic ssl --new-consumer --consumer.config client-ssl.properties contents of client-ssl.properties --------------------------------- security.protocol=SSL ssl.truststore.location=/usr/hdp/2.5.3.0-37/confluent-3.2.2/kafkaSSL/kafka.client.truststore.jks ssl.truststore.password=<passwd> ssl.keystore.location=/usr/hdp/2.5.3.0-37/confluent-3.2.2/kafkaSSL/kafka.client.keystore.jks ssl.keystore.password=<passwd> ssl.key.password=<passwd> This is working fine, and consumer is able to consumer messages produced by the producer. On Node5, i've another Kafka instance, i start another console-consumer from Node5, for consuming the data from console-producer in node4 To enable SSL (between client on Node5 & Broker on Node4), Steps on Node 4 -> 1) create kafka.client05.keystore.jks (set the CN -> Node5) keytool -keystore kafka.client05.keystore.jks -alias localhost -validity 365 -genkey - Export certificate from keystore keytool -keystore kafka.client05.keystore.jks -alias localhost -certreq -file cert05-file - Get certificate signed by CA openssl x509 -req -CA ca-cert -CAkey ca-key -in cert05-file -out cert05-signed -days 365 -CAcreateserial -passin pass:<passwd> - Import the CA Certificate & Signed certificate into the keystore keytool -keystore kafka.client05.keystore.jks -alias CARoot -import -file ca-cert keytool -keystore kafka.client05.keystore.jks -alias localhost -import -file cert05-signed Add the generated CA to the client's truststore keytool -keystore kafka.client05.truststore.jks -alias CARoot -import -file ca-cert On Node5, i do the same steps above,and start the console-consumer (to read data produced by console-producer on node4) $CONFLUENT1/bin/kafka-console-consumer --bootstrap-server node4:9192 --topic ssl --new-consumer --consumer.config client-ssl.properties This is not working.. what do i need to make this work ? ... View more