Hello Friends,   We have a upcoming project and for that I am learning Spark Streaming (with focus on pyspark).   So far I have completed few simple case studies from online. But I am stuck with 2 scenarios and they are described below.   I shall be highly obliged if you guys kindly share your thoughts or guide me to any web page for help on solution.   1. Writing Streaming Aggregation to File   # spark-submit --master local[*] /home/training/santanu/ssfs_2.py from pyspark.sql import SparkSession from pyspark.sql.types import StructType,StructField,StringType,IntegerType spark = SparkSession.builder.appName("FileStream_Sink_Group").getOrCreate() source = "file:///home/training/santanu/logs" tgt = "file:///home/training/santanu/hdfs" chk = "file:///home/training/santanu/checkpoint" schema1 = StructType([StructField('agent',StringType()),StructField('code',IntegerType())]) df1 = spark.readStream.csv(source,schema=schema1,sep=",") df2 = df1.filter("code > 300").select("agent").groupBy("agent").count() df3 = df2.select("agent","count").withColumnRenamed("count","group_count") query = df3.writeStream.format("csv").option("path",tgt).option("checkpointLocation",chk).start() # Error query.awaitTermination() spark.stop()   Error message : # AnalysisException: Append output mode not supported when there are streaming aggregations on DataFrames without watermark;   2. Reading from Kafka (Consumer) using Streaming   # spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 /home/training/santanu/sskr.py from pyspark.sql import SparkSession spark = SparkSession.builder.appName("Kafka_Consumer").getOrCreate() df = spark.readStream.format("kafka").option("kafka.bootstrap.servers","localhost:9092").option("subscribe","MyTopic_1").load() print(type(df)) # <class 'pyspark.sql.dataframe.DataFrame'> df.printSchema() # printing schema hierarchy query = df.selectExpr("CAST(value AS STRING)").writeStream.format("console").start() # Error query.awaitTermination() spark.stop()   Error message : # NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)   Please Help.   Thanking you   Santanu Ghosh             ... View more

Hello Friends, We have a upcoming project and for that I am learning Spark Streaming (with focus on Structured Streaming). So far I have completed few simple case studies from online. But I am stuck with 2 scenarios and they are described below. I shall be highly obliged if you guys kindly share your thoughts or guide me to any web page for help on solution. 1. Writing Streaming Aggregation to File # spark-submit --master local[*] /home/training/santanu/ssfs_2.py from pyspark.sql import SparkSession from pyspark.sql.types import StructType,StructField,StringType,IntegerType spark = SparkSession.builder.appName("FileStream_Sink_Group").getOrCreate() source = "file:///home/training/santanu/logs" tgt = "file:///home/training/santanu/hdfs" chk = "file:///home/training/santanu/checkpoint" schema1 = StructType([StructField('agent',StringType()),StructField('code',IntegerType())]) df1 = spark.readStream.csv(source,schema=schema1,sep=",") df2 = df1.filter("code > 300").select("agent").groupBy("agent").count() df3 = df2.select("agent","count").withColumnRenamed("count","group_count") query = df3.writeStream.format("csv").option("path",tgt).option("checkpointLocation",chk).start() # Error query.awaitTermination() spark.stop() Error I am getting : # Append output mode not supported when there are streaming aggregations on DataFrames without watermark; 2. Reading from Kafka (Consumer) using Streaming # spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 /home/training/santanu/sskr.py from pyspark.sql import SparkSession spark = SparkSession.builder.appName("Kafka_Consumer").getOrCreate() df = spark.readStream.format("kafka").option("kafka.bootstrap.servers","localhost:9092").option("subscribe","MyTopic_1").load() print(type(df)) # <class 'pyspark.sql.dataframe.DataFrame'> df.printSchema() # printing schema hierarchy query = df.selectExpr("CAST(value AS STRING)").writeStream.format("console").start() # Error query.awaitTermination() spark.stop() Error I am getting : # NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;) Please Help. Thanking you Santanu Ghosh ... View more

Hello Friends, I have just read the news that Cloudera and Hortonworks have merged as one company. I was both shocked and surprised. But now I am eager to know what will be the future of Hadoop, which is still open source and free software under apache ? Also, the Hortonworks Certifications such as HDPCD and others, will they still hold any value at industry level ? Please do let me know your thoughts and ideas. Thanking you Santanu ... View more

Hi Friends,   I need one help regarding Avro file processing using Flume and Kafka. In short, I am reading a json file, using interceptor and selector splitting specific values into avro sink, and then reading that avro source to write to hdfs as avro file.   The flume configurations I am using are given below.   The problem I am facing is, the avro file is getting written to hdfs with only header and body. So by using "java -jar avro-tools-1.8.2.jar getschema" option I am not getting desired schema of the avro file. Every time it is showing only "header" and "body" as 2 fields.   Please suggest how to resolve this problem. It is very urgent. 1. flume-ng agent --name ta1 --conf conf --conf-file /home/cloudera/santanu/flume_interceptor_multiplexing.conf -Dflume.root.logger=DEBUG,console ## Flume Agent with Json Source , Kafka and Memory Channels , Avro Sink ta1.sources = twitter ta1.sinks = avrofile ta1.channels = memchannel kafkachannel ## Properties ## Sources ta1.sources.twitter.type = exec ta1.sources.twitter.command = tail -F /home/cloudera/workspace/Cloudera_Share/Bigdata.json ## Sinks ta1.sinks.avrofile.type = avro ta1.sinks.avrofile.hostname = 192.XXX.x.x ta1.sinks.avrofile.port = 4141 ## Channel 1 ta1.channels.memchannel.type = memory ta1.channels.memchannel.capacity = 5000 ta1.channels.memchannel.transactionCapacity = 500 ## Channel 2 ta1.channels.kafkachannel.type = org.apache.flume.channel.kafka.KafkaChannel ta1.channels.kafkachannel.kafka.bootstrap.servers = 192.XXX.x.x:9092 ta1.channels.kafkachannel.kafka.topic = MyTopic_1 ta1.channels.kafkachannel.kafka.consumer.group.id = my_group ## Interceptor ta1.sources.twitter.interceptors = i1 ta1.sources.twitter.interceptors.i1.type = regex_extractor ta1.sources.twitter.interceptors.i1.regex = (?i)(Python|Java|Scala|Perl|Sqoop|Flume|Kafka|Hive|Spark|Jethro|NoSQL) ta1.sources.twitter.interceptors.i1.serializers = s1 ta1.sources.twitter.interceptors.i1.serializers.s1.name = BigData ## Source Selector ta1.sources.twitter.selector.type = multiplexing ta1.sources.twitter.selector.header = BigData ta1.sources.twitter.selector.mapping.Python = kafkachannel ta1.sources.twitter.selector.mapping.Java = kafkachannel ta1.sources.twitter.selector.mapping.Scala = kafkachannel ta1.sources.twitter.selector.mapping.Perl = kafkachannel ta1.sources.twitter.selector.mapping.Sqoop = memchannel ta1.sources.twitter.selector.mapping.Flume = memchannel ta1.sources.twitter.selector.mapping.Kafka = memchannel ta1.sources.twitter.selector.mapping.Hive = memchannel ta1.sources.twitter.selector.mapping.Spark = memchannel ta1.sources.twitter.selector.mapping.Jethro = memchannel ta1.sources.twitter.selector.mapping.NoSQL = memchannel ## Mapping ta1.sources.twitter.channels = kafkachannel memchannel ta1.sinks.avrofile.channel = memchannel 2. flume-ng agent --name ta2 --conf conf --conf-file /home/cloudera/santanu/flume_avro_hdfs.conf -Dflume.root.logger=DEBUG,console ## Flume Agent with Avro Source and HDFS Sink ta2.sources = avrofile ta2.sinks = hdfsfile ta2.channels = memchannel ## Properties ## Source ta2.sources.avrofile.type = avro ta2.sources.avrofile.bind = 192.XXX.x.x ta2.sources.avrofile.port = 4141 ## Sink ta2.sinks.hdfsfile.type = hdfs ta2.sinks.hdfsfile.hdfs.path = /user/cloudera/flume_avro ta2.sinks.hdfsfile.hdfs.filePrefix = Hadoop ta2.sinks.hdfsfile.hdfs.fileSuffix = .avro ta2.sinks.hdfsfile.hdfs.fileType = DataStream ta2.sinks.hdfsfile.hdfs.writeFormat = Text ta2.sinks.hdfsfile.hdfs.rollInterval = 5 ta2.sinks.hdfsfile.serializer = avro_event ta2.sinks.hdfsfile.compressionCodec = snappy ## Channel ta2.channels.memchannel.type = memory ta2.channels.memchannel.capacity = 5000 ta2.channels.memchannel.transactionCapacity = 500 ## Interceptor ta1.sources.avrofile.interceptors = i2 ta1.sources.avrofile.interceptors.i2.type = remove_header ta1.sources.avrofile.interceptors.i2.withName = BigData ## Mapping ta2.sources.avrofile.channels = memchannel ta2.sinks.hdfsfile.channel = memchannel 3. Avro File in HDFS with Header and Body as 2 columns             Thanking you Santanu ... View more

Hello Friends,   I have CDH 5.10 VM installed on my computer.   I have modified Network Adapter2 as Bridged and Adapter3 as Host-only, with promiscuous mode "Allow VM". Now, after starting VM, when typing "nslookup   www.google.com" on terminal, it is showing below message. Even I can ping that IP address successfully.   Non-authoritative answer: Name:   www.google.com Addresses: 216.58.197.36   But from fire-fox browser when I am trying to open   www.google.com   , it is shwoing "Unable to connect".   Similarly, I am getting response for "nslookup stream.twitter.com" and able to ping that IP address.   But when trying to connect from Flume Twitter Source, it is showing "Unable to resolve address".   Please let me know how to resolve this problem so that from CDH VM I can access internet.   Thanking you Santanu ... View more

Hello Friends,   I have CDH 5.10 VM installed on my computer. I have modified Network Adapter2 as "Bridged" with promiscuous mode "Allow All". Now, after starting VM, when typing "nslookup www.google.com" on terminal, it is showing below message. Even I can ping that IP address "216.58.197.36" successfully.   Non-authoritative answer: Name: www.google.com Addresses: 216.58.197.36   But from fire-fox browser when I am trying to open www.google.com , it is shwoing "Unable to connect".   Please let me know how to resolve this problem so that from CDH VM I can access internet.   Thanking you Santanu ... View more

Hi @tbroas , @asrivastava , I am reaching out to you guys in need of help. I was giving HDPCD exam one more time. My confirmation code was 34E-D22 and candidate id 8732865098. Within 15 mins of exam, the exam window stopped responding. I refreshed the window, but it did not start. Also, that time there was no online agent from PSI exam support whom I could reach. So I lost that attempt. I tried calling to numbers +1-888-504-9178 and +1-702-904-7342 for additional support. I waited there for more than 30 mins, but no one picked up my phone. Hence I sent email to psi support and hortonworks zendesk support. My Ticket# 13053. Please look into this matter. I do not want to loose my money without completing my exam. Please arrange something, so that I can give that exam again. Thanking you Santanu ... View more

Hi Friends, On 31-Mar-2018, I gave HDPCD exam. At the end I was informed by the proctor that result will be intimated within 5 business days. It's almost 7 working days now, still I am waiting for the result. I have already written to certification@hortonworks.com and my request number is #12624. But I have not heard from HW support team since. Please let me know whether they have any separate email id, or any Customer Care Contact number where I can call. Thanking you Santanu ... View more