8.prepare_streaming_json.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#write to Hbase preparation\n",
    "\n",
    "#from pyspark.streaming.kafka import KafkaUtils\n",
    "\n",
    "KAFKA_TOPIC  = \"IZkafkatopic.public.testiz\"\n",
    "KAFKA_BOOTSTRAP_SERVERS_CONS = hname2\n",
    "n_secs=30\n",
    "\n",
    "customparksession.sparkContext.setLogLevel(\"WARN\")\n",
    "ssc = StreamingContext(customparksession.sparkContext,n_secs)\n",
    "\n",
    "\n",
    "####kafka direct stream to connect to kafka broker\n",
    "#{'metadata.broker.list':KAFKA_BOOTSTRAP_SERVERS_CONS,\n",
    "#directkafkastream = KafkaUtils.createDirectStream(ssc,[KAFKA_TOPIC], \n",
    "#                                            {'bootstrap.servers':KAFKA_BOOTSTRAP_SERVERS_CONS,\n",
    "#                                             #'startingOffsets':'earliest',\n",
    "#                                             'auto.offset.reset':'smallest',\n",
    "#                                             \"max.poll.records\":'100',\n",
    "#                                             #'fetch.message.max.bytes':'15728640'\n",
    "#                                            })\n",
    "#fromOffsets=None,\n",
    "#keyDecoder=utf8_decoder, valueDecoder=utf8_decoder,\n",
    "#messageHandler=None\n",
    "\n",
    "#from common.serializer import deserialize\n",
    "#directkafkastream = KafkaUtils.createStream(ssc,KAFKA_BOOTSTRAP_SERVERS_CONS\n",
    "#                                            ,'spark-streaming'\n",
    "#                                            #,{\"group.id\":\"1\"}\n",
    "#                                            ,{KAFKA_TOPIC:0}\n",
    "#                                            #,{\"auto.offset.reset\": \"smallest\"}\n",
    "#                                            ,{\"auto.offset.reset\":'earliest'}\n",
    "#                                             ,keyDecoder=lambda x: x\n",
    "#                                             ,valueDecoder=lambda m: json.loads(m.decode('utf-8'))\n",
    "#                                            #, valueDecoder=deserialize\n",
    "#                                                )\n",
    "\n",
    "\n",
    "offsetRanges = []\n",
    "\n",
    "def storeOffsetRanges(rdd):\n",
    "    global offsetRanges\n",
    "    offsetRanges = rdd.offsetRanges()\n",
    "    return rdd\n",
    "\n",
    "def printOffsetRanges(rdd):\n",
    "     for o in offsetRanges:\n",
    "            #print(o.topic, o.partition, o.offset, o.timestamp,o.timestamp_type)\n",
    "             print (o.topic, o.partition, o.fromOffset, o.untilOffset)\n",
    "\n",
    "\n",
    "#directkafkastream \\\n",
    "#.transform(storeOffsetRanges) \\\n",
    "#.foreachRDD(printOffsetRanges)\n",
    "\n",
    "testingdfparsed = directkafkastream.map(lambda x: json.loads(x[1]))  \n",
    "\n",
    "testingdfparsed.pprint() \n",
    "\n",
    "#testingdfparsed = directkafkastream.map(lambda v: json.loads(v.decode('utf-8')[1]))\n",
    "#testingdfparsed\n",
    "#dstream = testingdfparsed.map(lambda v: v['key']['value_str'])\n",
    "#x_counts = dstream.countByValue()\n",
    "#x_counts.pprint()\n",
    "#counts=testingdfparsed.count()\n",
    "#counts.pprint()\n",
    "\n",
    "#testingdfparsed.count().map(lambda x:'s in this batch: %s' % x).pprint()\n",
    "#directkafkastream.map(lambda x:'s in this batch: %s' % x).pprint()\n",
    "\n",
    "\n",
    "#if using spark Cassandra\n",
    "#lines = directkafkastream.map(lambda x: x[1])\n",
    "#    counts=lines.count()\n",
    "#    counts.saveToCassandra(\"spark\", \"count\")\n",
    "#    counts.pprint()\n",
    "#    ssc.start()\n",
    "#    ssc.awaitTermination()\n",
    "\n",
    "ssc.start()\n",
    "#ssc.awaitTermination()\n",
    "#ssc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#customparksession.stop()\n",
    "#ssc.stop()\n",
    "#customparksession.sparkContext.stop()\n",
    "#ssc.stop()\n",
    "\n",
    "#ssc.start()\n",
    "#ssc.awaitTermination()\n",
    "#ssc.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8.5 64-bit ('ProgramData': virtualenv)",
   "name": "python385jvsc74a57bd0b3ba2566441a7c06988d0923437866b63cedc61552a5af99d1f4fb67d367b25f"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}