kafka_streaming_CT_img_consumer.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using TensorFlow backend.\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "import cv2\n",
    "import pandas as pd\n",
    "import os\n",
    "import math\n",
    "import keras\n",
    "import skimage.io"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Setting newly assigned partitions: [imgtest01-0]\n",
      "msgValue_type <class 'bytes'>\n",
      "[ -520103681  1179258880    16795209 ... -1969084278   680174120\n",
      "  -637564254]\n",
      "np.frombuffer type <class 'numpy.ndarray'>\n",
      "[[0 0 0 ... 0 0 0]\n",
      " [0 0 0 ... 0 0 0]\n",
      " [0 0 0 ... 0 0 0]\n",
      " ...\n",
      " [0 0 0 ... 0 0 0]\n",
      " [0 0 0 ... 0 0 0]\n",
      " [0 0 0 ... 0 0 0]]\n",
      "imdecode type <class 'numpy.ndarray'>\n",
      "Revoking previously assigned partitions: [imgtest01-0]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Aborted by user\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "# http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "#\n",
    "\n",
    "from confluent_kafka import Consumer, KafkaException, KafkaError\n",
    "import sys\n",
    "import os\n",
    "import numpy as np\n",
    "import cv2\n",
    "import pandas as pd\n",
    "import os\n",
    "import math\n",
    "import keras\n",
    "import skimage.io\n",
    "\n",
    "\n",
    "\n",
    "UPLOAD_PATH = 'retrieve/'\n",
    "#APP_ROOT = os.path.dirname(os.path.abspath(__file__))\n",
    "dirname = 'C:/Users/Big data/iii_report_HospitalStudy/iii_kafka_ETL_streaming'\n",
    "UPLOAD_FOLDER = os.path.join(dirname, UPLOAD_PATH)\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "# 用來接收從Consumer instance發出的error訊息\n",
    "def error_cb(err):\n",
    "    print('Error: %s' % err)\n",
    "\n",
    "\n",
    "# 轉換msgKey或msgValue成為utf-8的字串\n",
    "def try_decode_utf8(data):\n",
    "    if data:\n",
    "        return data.decode('utf-8')\n",
    "    else:\n",
    "        return None\n",
    "\n",
    "\n",
    "# 當發生Re-balance時, 如果有partition被assign時被呼叫\n",
    "def print_assignment(consumer, partitions):\n",
    "    result = '[{}]'.format(','.join([p.topic + '-' + str(p.partition) for p in partitions]))\n",
    "    print('Setting newly assigned partitions:', result)\n",
    "\n",
    "\n",
    "# 當發生Re-balance時, 之前被assigned的partition會被移除\n",
    "def print_revoke(consumer, partitions):\n",
    "    result = '[{}]'.format(','.join([p.topic + '-' + str(p.partition) for p in partitions]))\n",
    "    print('Revoking previously assigned partitions: ' + result)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 步驟1.設定要連線到Kafka集群的相關設定\n",
    "    # Consumer configuration\n",
    "    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md\n",
    "    props = {\n",
    "        'bootstrap.servers': '10.120.14.120:9092',         # Kafka集群在那裡? (置換成要連接的Kafka集群)\n",
    "        'group.id': 'images',                             # ConsumerGroup的名稱 (置換成你/妳的學員ID)\n",
    "        'auto.offset.reset': 'earliest',               # Offset從最前面開始\n",
    "        'error_cb': error_cb                           # 設定接收error訊息的callback函數\n",
    "    }\n",
    "\n",
    "    # 步驟2. 產生一個Kafka的Consumer的實例\n",
    "    consumer = Consumer(props)\n",
    "    # 步驟3. 指定想要訂閱訊息的topic名稱\n",
    "    topicName = 'imgtest01'\n",
    "    # 步驟4. 讓Consumer向Kafka集群訂閱指定的topic\n",
    "    consumer.subscribe([topicName], on_assign=print_assignment, on_revoke=print_revoke)\n",
    "\n",
    "    # 步驟5. 持續的拉取Kafka有進來的訊息\n",
    "    try:\n",
    "        while True:\n",
    "            # 請求Kafka把新的訊息吐出來\n",
    "            records = consumer.consume(num_messages=500, timeout=1.0)  # 批次讀取\n",
    "            if records is None:\n",
    "                continue\n",
    "\n",
    "            for record in records:\n",
    "                # 檢查是否有錯誤\n",
    "                if record is None:\n",
    "                    continue\n",
    "                if record.error():\n",
    "                    # Error or event\n",
    "                    if record.error().code() == KafkaError._PARTITION_EOF:\n",
    "                        print('')\n",
    "                        # End of partition event\n",
    "                        # sys.stderr.write('%% %s [%d] reached end at offset %d\\n' %\n",
    "                        #                 (record.topic(), record.partition(), record.offset()))\n",
    "                    else:\n",
    "                        # Error\n",
    "                        raise KafkaException(record.error())\n",
    "                else:\n",
    "                    # ** 在這裡進行商業邏輯與訊息處理 **\n",
    "                    # 取出相關的metadata\n",
    "                    topic = record.topic()\n",
    "                    partition = record.partition()\n",
    "                    offset = record.offset()\n",
    "                    timestamp = record.timestamp()\n",
    "                    # 取出msgKey與msgValue\n",
    "                    msgKey = record.key()\n",
    "                    msgValue = record.value()\n",
    "                    #print('msgValue',msgValue)\n",
    "                    print('msgValue_type',type(msgValue))\n",
    "                    '''\n",
    "                    with open(UPLOAD_FOLDER + str(msgKey),'wb',encoding='utf-8') as w:\n",
    "                        w.write(msgValue)\n",
    "                   '''\n",
    "                    \n",
    "                    #imge = np.asarray(bytearray(msgValue), dtype=\"uint8\")\n",
    "                    #print('imge_shape',imge.shape)\n",
    "                    #print('image_type',type(imge))\n",
    "                    #print('image',imge)\n",
    "                    imge = np.frombuffer(msgValue,dtype=int)\n",
    "                    print(imge)\n",
    "                    print('np.frombuffer type',type(imge))\n",
    "                    img_decode = cv2.imdecode(imge,-1)\n",
    "                    print(img_decode)\n",
    "                    print('imdecode type',type(img_decode))\n",
    "\n",
    "\n",
    "                    \n",
    "                    # 秀出metadata與 msgValue訊息\n",
    "                    #print('%s-%d-%d : (%s ,%s, %s)\\n' % (topic, partition, offset, timestamp, msgKey , msgValue))\n",
    "    except KeyboardInterrupt as e:\n",
    "        sys.stderr.write('Aborted by user\\n')\n",
    "    except Exception as e:\n",
    "        sys.stderr.write(e)\n",
    "\n",
    "    finally:\n",
    "        # 步驟6.關掉Consumer實例的連線\n",
    "        consumer.close()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}