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 - 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

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

Hi All, I've Kafka 10 & am trying to view the Producer/Consumer metrics using JConsole... I'd a basic question regarding viewing the metrics on MBean tab on jconsole. ref -> https://community.hortonworks.com/articles/73897/monitor-kafka-producer-for-performance-1.html in the link -> Do i need to make any changes on MBean tab - to view parameters eg <strong>kafka.producer:type=ProducerRequestMetrics,name=ProducerRequestRateAndTimeMs,clientId=console-producer</strong> Or are these already configured when i specify a JMX_PORT & start services ? when i login to MBean, i see many of the metrics present. If i need to add additional metrics, where do i need to make change ? ... View more