「センサーをラズパイに接続し、MiNiFiでセンサーデータをTailする。そしてNiFiに転送して、NiFiでHDFS, Hiveに書き込む」の手順をご紹介します。
実現したいこと
- MiNiFiでセンサーデータを取得し、NiFiに転送する
- NiFiでセンサーデータをRawデータとしてKafka経由でHDFSに保存する
- NiFiでセンサーデータをHiveテーブルに保存する
- NiFiでデータを加工し、温度が閾値を超えたら、Slackアラート通知する
環境情報
- センサーをラズパイに接続
- ラズパイでセンサーデータを取るPythonスクリプト
- HDP3.1(Hadoop 3.1.1, Hive 3.1.0) & HDF 3.3.1(NiFi 1.8.0, Kafka 2.0.0)、クラスタは同一Ambariで管理
構成
構成は以下の様な感じになります。パース処理のところは一つしか書いてないですが、複数のパース処理の集合だと思ってください。
![enter image description here](https://lh3.googleusercontent.com/YntfXL0IrK1PG-D07WG5A6vvrUKY4rIGT5jha0uQEA7ptUT8bH1__4rahK0WJX5ni5eREZKYf9gJcOJsh9LSXDGr4yqPo7CxfPKkNS4LJTz8HGxoByhXiBL7h9oLoRYX-4-uI1zhMbPbqdARBQE7Aa3JwLQ-yVncQKFmrImcmR8ozGznOdVFwT5VjZhxFuiLuFypThTOph3x3ZOzwdUv_KeGHA3thWJbwSgdRCbOwh7sV8dN8e7njJL8Ouoohx4_5w7MtNQ4yw9Z4t2B9oGp3P9sBCiyOKSRm_95BEWsFk2BLab_u9nb3egSXHOcX3c2qCwsrZtfZQ6IUVMPQB8FOG3k6tvxnU_CFhzPYIZ1Fb6nvL5KKgKMs3Vk7p06w7AzMd9RH_NDzPn69QOlCZKqucDVYUcRXda3dMqRp6vs2iFyR5u529WCYGPyWp0jG3C4qfOfnJzD6N1ZHJioiOikcehg0Wx8eDYjEGFN9mdaQfOUafPaf7LizAtEgGtaiNCG-mfSqz2ESpdZzk6IYibb7iz-_uXxsHW15L1Zxd9LdZChoLXUeoIqq-g-LIcaTp53kJ0_PBvKTbmlg2CXpgyQkOZO8-0nOILYr77rDYZQcpsWK9IOYzt3U8M30yPTNplKcaF-FFYpwmiErKTnhKa0Lo3HCQ1Jvqc=w1552-h804-no)
全体データフロー
- RemoteMiNiFiというInput portとProcessDataというPrecess Groupで構成されています。
![enter image description here](https://lh3.googleusercontent.com/gAH0AHENaXp0fYl_A4DZdj-oZ3TrHilGrOjn69td6PBiyMPMrQEPfVzrGfq94x0MrP39UUnLUXYEf_jzEeB0WayTjyL8ZCii-R_QmLqgcMwpuMd7rBp2dLF0wiKgoGf3SoGckMQlhtYxF2S7CJynSJUTmEH9VW8eKD5M1UhwUvcD7MKGf94T9-o7s7j5dU0AJGMFXqcEUz11askxsu2EYbeUvHAQxb7vdzXeljTBxIywSRqvIiFqJqhC9sO8Bzq7dYdUOomHNX9uu7uZLQ4qTM2KgU0sFB99IscV1zRaITKK_BJrc9aZgE74K1_l3hLyN6bAx89lwcJm5XH_aHCGo0DS1tcnnFeSousRJXXIA0wWbG-ioBKtHOIxU2S-2gyA5FJBC53OrUa4Db1pVp3ozpBHfZYA_YOyKH4SFojVy-TA1n3tUWUXAQ4RSjQXP2lVAqgXugY5AHGH3JsBB2fi-x4h-zUk1LygY9kNMyVHX5afkG1O42xLRACFE174u5srgUtmpb1t8MRTf_eQ7cZpuOR0ONsgkAM1efoXwuyt6EXWU195G1CxpvagEGVmxXra3gF0X2ngZVAgVoxKoSwUgk7RQf2ZeJPAmo7yq8PqRN37beV7PPplD7cM6pdzxIqTgWj2Zq6H4DS9MuOvJGdKMm1SF1UcSRE=w1894-h610-no)
- ProcessDataグループ内の詳細フロー
![enter image description here](https://lh3.googleusercontent.com/QVa_PGqbxzJlywAGpEqBaDIHRFEpZvA6wGyUicyaA9XB_l-3QgBmgrOs9UTR5NgGUz7-qHPa4pIgD6XGCAIXjOUBCx46-RpdMTi7BaIQRFgjJg6_HaC9OM2tzTX7Z3w8YjUEr1wdwhj1kwoaD6NXOTA4pCginU9VQZN9_CNSHoIJHpYEQ38fgxOg0VZb-W1gpvgPBuGNYxpoStx6ZEVPQsmTHZkjwA9GlhQBe0rX6Fw78RWDOxj-DpCUIPOoLDjNk_LtWkcnp9HnAnGd5JKbp7AKcfqZK460f3yuyg9gfQuB6llLbfXZyrh4Y6d1x69_5mJh9KZVOboHU8at1vMSIkYElzFRhkTimlfW5mnrPmMrKM_9d5RwQRl3tmgCOtQMP6aquOH-5qNPIwfI8Wf17acV4R9C9TmAieyhzYLFyN1ZkVNVzQOorCSZy2oNXHaAZbuTsLaQhNTd35ZBmPfOAMy6R27W0Z0_1Bqnqp00k8OWknZEX16DoIaU1memf3AB82pV4ChKaPsfXHP14Dpi6Y5SWFaRE_0u2NGPlDV-LbVM_8yOZcGLc_N3qxoZJnWmbPKI-g10phQgQKOOYcEaerT6EaMr4vTgMhl9uaopSg3TJmOlrEqNFm3iTeK3nZ84QHttis2iy0foFKS0RaPCLGGuSVwRIqk=w2064-h1334-no)
やってみよう
1、MiNiFiのセットアップ
ラズパイ自体のOSインストールや、センサーとの接続ができた状態(ここでは割愛します)でMiNiFiのインストールをやります。
MiNiFiのTarファイルはHortonworksサイトにあります。
https://docs.hortonworks.com/HDPDocuments/HDF3/HDF-3.3.1/release-notes/content/hdf_repository_locations.html
sudo su -
cd /home/pi
#HortonworksのサイトからMiNiFiのtarファイルをダウンロードする
curl -O http://public-repo-1.hortonworks.com/HDF/3.3.1.0/minifi-0.6.0.3.3.1.0-10-bin.tar.gz
curl -O http://public-repo-1.hortonworks.com/HDF/3.3.1.0/minifi-toolkit-0.6.0.3.3.1.0-10-bin.tar.gz
#ファイル解答し、シンボリックリンクを作成する
tar -zxvf minifi-0.6.0.3.3.1.0-10-bin.tar.gz
tar -zxvf minifi-toolkit-0.6.0.3.3.1.0-10-bin.tar.gz
ln -s minifi-0.6.0.3.3.1.0-10 minifi
ln -s minifi-toolkit-0.6.0.3.3.1.0-10 tool_minifi
#MiNiFiをインストールする
./minifi/bin/minifi.sh install
#必要に応じてリモートNiFiホストを/etc/hostsに追加する
2、センサーデータをPythonスクリプトで取得
Pythonスクリプトは/home/pi/bme280-dataフォルダに2019-06-04.csvのようなファイルが生成されます。
中身は:「センサーID, 日付, 時間, 気圧, 温度, 湿度」のカンマ区切りのデータになります。
1,2019-06-04,23:59:51,1011.54,24.86,50.67
#coding: utf-8
import bme280_custom
import datetime
import os
dir_path = '/home/pi/bme280-data’
now = datetime.datetime.now()
filename = now.strftime('%Y-%m-%d’)
label = now.strftime('%H:%M:%S’)
csv = bme280_custom.readData()
if not os.path.exists('/home/pi/bme280-data’):
os.makedirs('/home/pi/bme280-data’)
f = open('/home/pi/bme280-data/'+filename+'.csv','a’)
f.write('1,'+filename +","+label+","+csv+"\n")
f.close()
このスクリプトをCronとかで定期的に実行するように設定する(例えば10秒ごとに実行)
3、NiFiでデータフローを作成し、MiNiFiに配布
MiNiFiで直接データフローを作成するのが難しいので、NiFiで作成して、テンプレート(.xml)としてエクスポート、MiNiFi toolkitで.xmlを.ymlに変換するのが一般的です。
今回はTailFileプロセッサとRemoteProcessGroupを使います。
- NiFiのTop画面でInput portを追加する。名前をRemoteMiNiFiに設定する
![enter image description here](https://lh3.googleusercontent.com/9o7pS1GfARbTkrwDmsukXH77B1Wge5owdPILVIBpcCwWKwpmRWxTPuC7pf3rS-DN_H9MDwTWcld91m2CcXnngZ37UK7cVMXVnOFba8-MuV01_kfP9-fwtJvogsjO7IMx-ryHWZqeQRFn3Qj9t_jMcjzV08zNbA5X5G6U15OqUgFH6MshyjfDWeYkIeLWS-4DxBcCeZ4m2UofEfiV49trzW7V6AClNEXCNuKDszz_6MTWaj6dXyUi5S164YlUWTTHL-M3o_ViSUABK95QNDBluGh14hVR3DFV3OVXTKVcHXd7fEiJNCk-_jzYmtWC8TM_guZCOeVIsrH7eLis9_eObFCSUERvn1ukUVe-jwFlH2OD3CSkwhP5CYrs_5vUd2pIotGfVfaO1hsiS25fZ7TnWGpnTQnyYv1cjV52gWJxbiVs-BExAJSkZROJK8CTu9IcdISuOY0PZfN311_IYg_3sTu9WEq91cC9AIoDdPlK_-cA27Ji8mb8hNkHN8AseFh-owBY5MUClVlUyJvg59gMMIZpgUgcH6r2ThM5rr3Uwgq-LeibKE5hpGvshyRPYXBtnzHrxav9odxs58oslH7Kq0pNDqBWTMmsEBq6idIGmBjmbRpra7tIAn6fyZiWMUDlmkqASUO9pSVCu0sSNVyYG6wGIVazZJs=w682-h482-no)
- TailFileプロセッサでフォルダの.csvファイルを取り込みます。以下のプロパティを設定します。
Tailing mode: Multiple files
File(s) to Tail: .*.csv
Base Directory: /home/pi/bme280-data
![enter image description here](https://lh3.googleusercontent.com/HjMFeIkC4HloHXFOahbzF8xEWRZbTArwEdGCSWWCTOuFvE0L7-nq6EhzuBitJdiYuBnq2hClYhUORvPxvhmupnMpJn7fFHjIZzAZgCVkj3csVIzUjon38zY_zZRYQFy40_R9sCdtpBGtbi-cAO5pJeN_W1FX9TgPVpIrj5vLghnMwGsRTWGBMmJHG9BZyaCLJdxvA2w-AKFwYoz8g7ZFQ0H2xAH7COOunGUP3v1n67zYE8SL9IIPWQREFTh4KBQUECFl4VJ4YTTFlc57q4aa5uusIx_BGqjf9C_nQ57RGNiBMJOuZ00M0J8G61P4NTTRnRXMdj-9cAgz44ffH2s0jwrgAyLlzvTbX6Gu8oe3dvj37bj_25VPt5E99wM6dGbquVNymRoluKbuMMiV1OQIEdxCM9DdncWoODxNKpjzHfsgFoA4VPr6_qficdUf1-B-ei2j8lZFll6Z0ZOd_kPP7RUH4SC5Wf4YWOtPzTS9zQXz9gIh_DutvWhz0PWQ8mUrDjlSfuNqgtDrh0QQ8Azfrta-llmx5vIBQ8LdPMqIfRbZlS4i5AU9yQZUR5FFLsnjGdiSJ7CyCsCJ6m46FZPY22Mo5qGedOtYYORR1rzuoruoGO3tJr8O39wku4iFBaFqiRzrdmOCgzDop_xkZZvCZ6eNUpUtRHE=w1568-h1118-no)
- Remote Process Groupを以下のように設定して、RemoteMiNiFiをInput Portとして選択する
URLsにNiFiのURL(http://hdp-srv4.demotest.com:9090/nifi
)を指定する。複数ある場合は、カンマ区切りで入力する。
Transport ProtocolでHTTPを選択する。
![enter image description here](https://lh3.googleusercontent.com/PVXdbZLTOjZqLJte44i6YNRv9lk1MYdDjPbeUIfc8UJOYapuJedjzvWhGwd2C3hilmVznGJxj8iQGvDl1zebLgs0A-yxO0-axUw8FaO-ptM4Be1AcH9-0Juxbv5Sj0Iwz3-uC-xRI_bifgYYM6am6xyZEdiWsGry6sqHniEB3v7Jy2wegD87tFluVGMHP1FNJsjBFvzJ6dbz0YQp2Gfw8cGT47-i-q6y4BHmsMUJkLz1XAeYC8pxZYGtGFbfVIQFtBd1KLpY6xzL_uVL770YQynkCM25eZ8Iv2c6rc7b9bWQQR8zJYEQcWkIeJpRFbIQxbaaNhTB0rW8sLjiTrtrM0C9S4wX64hQU59SDJw733MxcpRpT8z27-VW2igpUFhqZF0ERKVW8EQh27hszQSoxkmtzwnb9c30Xz_OUEBzNyixkUB12-p6WWvyfDlOUkPGcUzKtqrt6r1guc-_NWOnce0n8oRGzBLmM6SiVhrmwXIwc15jZsqk5FiRQ6mniTgfrT8DEhDFQF5wDtyayQQvwAqH9TPNQZzWUTlbCmtJbjjMqrRAIwa5M-IEbq3vCaPhj3Gt6duRrd8NldeRJT0fJs8jxCmIXZ-XwyLnP82k4VU4cGswSM6h-itT-SCCpSYiRKr5ohlzjL2xLA29V8b4m_AXVwGthiA=w1508-h1118-no)
- 作ったデータフローをテンプレートにエクスポートする
TailFileプロセッサとRemoteProcessGroupを繋いで、全部選択して、右クリック、「Create template」でテンプレートを作成する。
作成したら、ダウンロードする
![enter image description here](https://lh3.googleusercontent.com/jj5w8z33gg7lK-CKvV-TPgYaZW5-wXpbRHzd5IPsD3IhJ0017uC5RW54awerFEIzhNSx_CxbN3JEUONG2g8GDFbf6V_3_vAMXcO4EZQJuBtU8io_n1wGf29KIqqkw8wOpcDNXwwbEOWffffoa6QPJAnu-40yt5anQ3j2DzOwNF7g6FBzDjCwIc9J5PDQqbF1STxI_KhDfAp5c2n6S-eIC9w2AyLYWTZxIpziDSqSHpmCePCq-_I6WMjjOzGK2vM0_nmE_jx9ACTVD-gZmgFQGRrzkQuUnuxoC5k57N4DN6uHkUxhNq06DmpdJKwCo1htfKI9bHCDusILYlxQfj3H3zb93EDK5bXYKtzQwroKcwP4DvsZAnVZoSJxZ3W2uzyJaKvoqR1kAF_NrDQfSmUmgLMgFvNvrildQRsK5oAOK6IiQ6O04E4LF3LHBsaGHEa2RmlLTzlkPkqQ_oe-3QIDIM7Aizt7KetZxg0GyyyCNJlkYLwOj7aokJfbt8554RCzTQUpklp9ufvSwHwyZjxgKngkR7Xi8hTMbn32jGN-rNkykmuWqXL3Y4u5kRZGMy45dWI_bnxmbrPurgLK8MN1FjFmvRPruaGJ0ZKljP-wZAhL7VVpkXZnbPkt4FwCfdKYll7iVS9VbT0oLa2Txemq231TsmeLjks=w924-h802-no)
- MiNiFiでテンプレートファイル(.xml)から.ymlファイルに変換する
ラズパイにログインして、以下のコマンドでファイルを変換して、MiNiFiを起動する
sudo su -
cd /home/pi
#MiNiFi toolkitでxmlからymlに変換する
./tool_minifi/bin/config.sh transform /home/pi/sensor_minifi4.xml ./sensor_minifi4.yml
#既存のconfig.ymlファイルをバックアップし、新しいymlで上書きする
cp -p minifi/conf/config.yml minifi/conf/config.yml.bk
cp -p sensor_minifi4.yml minifi/conf/config.yml
#MiNiFiプロセスを起動する。他にstop, restartなどオプションがある
./minifi/bin/minifi.sh start
ここまできたら、MiNiFiからセンサーデータをNiFiの方に転送できるようになります。
4、NiFiでセンサーデータをRawデータとしてKafka経由でHDFSに保存する
ここからは、NiFiでセンサーデータをKafka経由でHDFSに保存するデータフローを作成していきます。
全体のデータフローはこんな感じです。
![enter image description here](https://lh3.googleusercontent.com/kkMFO_5e-Rbfy65vOg3RXITIFet-Un7Tc_tVCyPh-jXU2JerEM7pr4NtCzYt-l9c9QWudn2p2iSuBcZ3AsVtS457vyduyazObyVkdYKBIbdY2X3MUAun97cU1ZqRNOO_OE_Jp7luEqJOkpELYnFEFmU4Acfc8p1sS6kz2iu1i0SgasIs5n8oB1WlnsIN5We-qxjPNEKqHidJ3q6kroZznhwCZ2f0Bf5PnKSjMdGPfFiSlnmBy2tJOsxeLDaY1rN1YN_nwnccLVeBgt1UAnZBKi_2XSqQtvdf6Y2gBZyoH36EShkRrNF3hh0w7xcwMSTwLBKVb-doUM2L41Tlz7KfDaDiCsG_9JfcYRLLsgeUa_H-86Q4gWqTPbWBPq_5BzLGjXrAXCA2kfCGAhp5Y4DFt_hWoJE-do0VfPezmTfEAst_H8pA6ldrQW6AfEaWBooWlb_iM5P7mdZfsIRUwzIhj-FbX-unj-jILfJmxYoMfZGypbxq_lLpDwTta23viWQjFyEcsxd45lA9_oexr007NuQjVRGCM3_YGaBNx11GCskFfPjh044byAhiZN3WUO_f-Te0Citlu54QR6YSXnjcbygPyCtbnhGPF-KAuFCUISUnXkPLE1g86eaPadRlKDEErePKD-xa7Obc75SS7iIKFUODSQ-Y9GM=w1784-h1494-no)
-
NiFi画面でUser1というProcess Groupをドラッグ&ドロップする。すでに作成済みのRemoteMiNiFi input portと繋ぐ
![enter image description here](https://lh3.googleusercontent.com/gq6EKenOXOTgwHmY7a5sSEk7mzuOAoSVV7Jxomx-XPGxreIbeuYHotsIHtXjwgoaYz02gIW9O19j3BGxs5w8LHMz0SjamK4dtDNxTcDQW4ZJPMXasnP9fmOt3lRRSWatz9Bg-dLkRtyKeq3AddjDpJBxiz382EREhHJ5_Gk0gNGzuQ_1tZrtB6eyiuJyxudUsEYY2E3KHPKeYXe_D3-aNY5GGUmylB5bR4vrHrQqrHZAmKg5rUbDCv1px-nbjjTJ6i_TqfXQkdXHQbVNuhWGPiJbntMR__XhLNrz_kHkpNr76q2XjiUpN8vO2fkSsZ25dru32TI94u9us-nZqgO38a9hwoT0QFQOW-Pn2HbcsbpbfJqicPm7kwStl32y6hpdzxeJeYWRSWEHrkM5IklZ5046ecZZQtHIG3qCGxjFdwzxw2y9Wj2e5aRUvEu-Zc8nmrAEGp1zm69ZVXlf5j0ekBD8u4yaYUlYoeEZQCnTrjhGX_8vcn2jNixmyB-7rOZbVVboP0LL2McBGhp41G4w5q95jqcxlv2WV6Sw8JeDT-J-El8njouWxqdodcXb_SO9doMeCbl31Fh62aGgki2z6syk_Y_8tt2DS9t8wimND-JRR_AgjJerl_6GCzbNPSAZ-Kn_LyiEPR8EVWq_rQ1Sam743-2KHi0=w928-h1158-no)
User1グループをダブルクリックで入って後続のフロー作成に入ります。
-
FromMinifiというInput portとPublishKafka_2_0プロセッサを作成して、繋ぐ
![enter image description here](https://lh3.googleusercontent.com/8ess921rKQOPHGHyneL2gI1ZUYmSYbGcD_n2H23SgIFUpZw5lEzkYkKkyjIGYq-hzDULku5CZkZ4WrrKi2HMNEEScM_oNBGVVsvpVCMC1ON3ivLBSOJyklmXrXDMjncQMMnbCgF0GjZskKMvVK7NwfMGMPVzREe7cz7mNMwPXkXzpRH7oh34roS0UiiG_NiKJ7t_UE6roAIAOH3fyjtbMD2ws0l8Cm06Zg6_Aupp49anXShynoBPZC9gVCZpEUyeNHN-LNzPJDTX7nzcGzndTWCJIAbQsGus8eHcdlFvKcjBT7nlMPw6X_06KYtor6EirfbGtiWziqneIzNCtTplGxEb7WWd0N2U8Ste7w22iRJuR-TZa-SzpMzzznUtPZF6PTH8tuTnWY37n4WFBT3XJXw0nxw-Ui-01b8niTPAS0SA9ecT8FosaAyvAqIAFvZG8gKR5JYlk7htUycCMwSab6_iUUiKwhWh2r84PDgfcEZRLe10qyu27IesHgN44KryU9Pr9aQfqEwoLwV3IWZijb6G6-EkMYBBdtMnGeT-YkvABymIE8D4q5WOkliy9OA9HNT-aKKtiu_2NoDwNbMEUAJZRwFtSOGlhh4VDDGqJYZ6Zj9m0YXjamQBSvXNQIKUcgdXTtI5NrtIjWNdxnsIR7PT991P4ss=w1022-h744-no)
PublishKafka_2_0のPROPERTIESタブで必要なプロパティを設定する。
Kafka Brokers: hdp-srv1.demotest.com:6667,hdp-srv2.demotest.com:6667,hdp-srv3.demotest.com:6667
を入力
Topic Name: sensor_data_user1
Delivery Guarantee: Guarantee Replicated Delivery を選択
![enter image description here](https://lh3.googleusercontent.com/IEdzR82eOSyXdOYnfLSCyp4HYQtxmvLGiJ5KR0TJuSurkB13156Q6hdK8eQbnhPJZEGT9EEQAe-P-wX9DKZTaqvlR8GmutA0KL4-3uWlz6Oqho-j58fY3cVlmvo2vmQWEo4UQJv5RqWqwmDwEbHJH0t5XB4VOlSFc8Ps8phMzIgUHHdsiejM0ozzu27Yv7bNOxbpKwESlkQFNU_wgnevl6NRgsZvUaH31QaDU16k1pK51oz3d07w_50gUsMtwG5j-veSywqDhrSnYWL0pPzSjhxwxet_x3YwoAhDCW0YZ7od0qqev-nUx9w4fDnk0bsV8EWJ5p7oWQ2lrodh4A7B1mwELNwMYhrEYpeFXgkoYLVhm4QoV90DxZ1XoTsjFBwfzsSzYOPOUe4U9_7CZoyoHHBfGPeeql5-iWGua31I9IUESOFpPMxQqr0GyTa2tcJnbdkF5Pc9fLEBr4Fhjl-sI61ZBZsOj3t7B6bE1Ind4PMTKsylV8yv6eQOKt9Ah_Hs6xJtb1xtXSWj31pUVcE_drcqDz2AODuFBZmJo9-nFOrAOeO-zkEgaapRKzH1J9kJP953dy4LhED82_HRSQg_g9CVP38kOzFOjyMAS2j-9X5PghS1Td_lh3uOmmKuBUser0XPqNrg6nww6b_qOXCcuzYogQiy8Q8=w1570-h1128-no)
PublishKafka_2_0のSETTINGSタブでfailureとsuccess両方をチェックする。
![enter image description here](https://lh3.googleusercontent.com/aBt1UkNeKZtzX067rRGYYfSFpME_tb7FWT0AsA1ZqWeYy4uU0-7WsA51zkOGG1i2BJLt4A9XyhqVxxoGCb9VORlT3oftm1c881ljeGa9I1NMiNPasjurssqcyipzQyBFrBabWNMH3iJiG8xzFfSiVz8K9BsTujeBBcAd5GQpLcM3CV8SHkHh-vwVVz-zFjUjuvZ0o1pSBu1SfZMNiTEyU7tsjAGiCnq0Y_hzzHrPojqSRXMtmNoGdthIvDJuaMmfOt64D2vaVTBPikptkmWyCM19jRA8efw9QRXHC2xy7a_uTQtySp1i4svTkbU9r5LaxOCPZvEryJ9Jq0__eghSeypP4RWaBoeGXnkHHyqvcX2KmpFGxnJV5sF6D1t30ktAS42P9_qSoWfVI64DzROyPI5QauR0S3ZIWZmi5RojqH5CagunNaF4J8rt5SP2qP48hOq1AsfmjWyUAaEgoSaS_udOEuOnvJ-KhRFXHbLul4-7ZaXNrgdMBNDmxFTpY4dtdW_ZNqWA0jNQ93yfeyoJGwuhNmXqu0Y18v0c9n6UMf7ALeQQ0VCF-8zablaAv43BH1RKM6pB-KhZt89ITqe8NqJ2qSCJ-22mLa4MLV7O7SGpmUUsO8qLZjLmJlVavpUoAx9CkUvxjgZANqTOXfMOKg9vCNlpHG0=w1578-h1138-no)
-
ConsumeKafka_2_0プロセッサを追加する
Kafka Brokers: hdp-srv1.demotest.com:6667,hdp-srv2.demotest.com:6667,hdp-srv3.demotest.com:6667
を入力
Topic Name: sensor_data_user1
Group ID: group1_user1
![enter image description here](https://lh3.googleusercontent.com/VQ9ZR0dQzw-srmwyGgHolUOqdcEHlgd1Fc2SvrQBLIo0rSFkabDVqTlbD9yc3FEj9Zqbsyn1qMyPI3nmwCaSmY5lh8rW64e8tRvctmMpJ59Tu0cOFH1wiYVUbwNia9ufKa7sjvxtHU4tXEDY-5j53FGqElFYIDLm5xHSS0sKdY_IntL3l5Y01wgiyVip7Fd7mdgbQJG4FBvtAS8v4xKqOae08vGprIriFkjhXE-r6rpKKzbP9Sv3slIcoAHTrsQINo-x0WnuUQ_0UWXs6A8bx_qCeuL4A6-LQB_gUPqt_n0KJvkmfYgIsZM2ndMO8NY8Vz9pld22MFd-nKQmixvn5iEsGZTQEz6k_Sf4R9u_mwmSB8nWh4HGX0kFDXDKxv_tws0wrNbF2bONIwnLbKX-YjFurhdFkNNhj-QnVF0sQboHeCchWjpsAQsvxfMaZE_MKKdhF-vgpp2S-1QqWt1C9IEBmCLjfB2rj5qgTPDaeolmKnRetq9WOe1afRGkEikPSpqeHmxp-AeMgoKG1uIEQK8Ptp10O306czoCvn366NdK2-9kxnQM-eb97DVBU2r0UtwwcR02DmY_eJrleUSS9vT3uSZjgpiylY69c3u1K_HHN67KaexWJ8hW9aEQhKMb5N-_Hg52-NyyICaVVsE_3nTFWlEet-E=w1572-h1128-no)
-
MergeContentプロセッサを追加する
Minimum Number of Entries: 10 に変更
![enter image description here](https://lh3.googleusercontent.com/BojIZFjnkEilWCyq_rzt5deYDCFLtCIsDzkWBM7GBVn4bpZknJe1u7W89x6VHCOrkYyKZkSt3NiBSNRHOEMFlUoTTKcN4If0qiF_mozUkXtXkNswlR-I6puK6Um1pgKIcC6o4HdeMVbKf2E6yBLHwqSdHQJ-JrzAuSNz2RZPzm6_WMk1kizfVLlVYUA02ZtlR11iYp2_CQkaFNPzQ7LeIZI-m82yAeEjeIAnmuHK4mXIkoJIiH74Qy5koMD8x05hTnGK7_4NeeelyskOaBV3L6-AV7vunAlhVP5loSaKFylQH0NzekNoL_p46IN0n0WPWr7-9vX7OReokjV6r9ig7bk_PsWnGrBOyUhKc86E3mf-8_4Zm9y8Yy9kmx1deW3KAjwJm7pTFoJIOMfeCxI7uAKDzkGx0cxpSOyr_K6FJskXrj2VUm-xDtatP4GMosXqE5LlMZSr6tRUtJdzjEUjXvGsE5Y__5Ex3UkvkjMhpQjsvxNMM2rdgVbYF4aPzEntviEIstGfrPUEyaMhNVwwJ_iy79Etu9LDUKjC2RTIAWgxvs6IlSzy0DwW3AROZKad7jcfc16-TNkbAw7kqHuMos9FqcUQocWz32JgguuUkkQh5YjmID5REnwUpA2kXs1O8ysrr76H72gm3KA12IYJS178kbW8w9k=w1574-h1128-no)
-
PutHDFSプロセッサを追加する
Hadoop Configuration Resources: /etc/hadoop/conf/core-site.xml,/etc/hadoop/conf/hdfs-site.xml
Directory: /tmp/sensor_data/user1
![enter image description here](https://lh3.googleusercontent.com/bET7tbjtvk-aLyL4SxqK-bLhkZycx2hxPI2c2QpmVM4E0ZfrU8CVMCTKgqvlNhMXj-Zq2BQEet9kLkXIOFN7O0bJ0WZdQ8CM7fpN_j40DyCSQ3XT3jzb1eWNlZoZUf1MSB6Q3LWm587RZWw9un2U5VvOj-3wkqAs9tSl9zWUI_tCsCXU_6S2G9bTIlw8F-Yy7WnQ-4SPYFcnjthUt0iKzusFwqeddVZ-_KHiNYJ5KXgLsDiAC1qKrPwSLzcIAp6O9Iui5OiWPg24uuLwo9ToexicdMwVkQN2sGs2chgZ0trFXGLcAKQgEGteguQMIKpxB2WEP5Y62iZld1FGsDJPqNQrbRZ1CDi2sM8guLJ2clftX5nHlr3SQ2627Y-4ZdCW3f3Ybc5my8CMudS6tMUN1wgO5823UL15mR9fQb4xZExnAY8lHAK8OlONsrDur-0wMfRS_fH2zJKJCwtr3wsecDFt530Txy1yKewgk1ebJuTTcRbi8D0NMRbO-uj8Hmk55-gOmVJ3zBikj5mcXk_wOatamjxxk5wM976xGUbh20bWcKlmeUH3nOd2d7rQnZlRdaYZlU3hKG4R6oIpXPk_9KdGpCojuez7ZNM1XWFyw6yM7ift35NjSBAzdy1iQwuLd4xD6PXBv_mNRDDDbrl7etrMW-P-F5s=w1572-h1134-no)
SETTINGSタブでsuccessをチェックする
![enter image description here](https://lh3.googleusercontent.com/eI8WDb5bW5orwIHEoSEkwpjrgAn6HXV4kX1w91BOwpmI4fBdLUKPPi55SQzAzyf852NxfVYZGqb_XmBfzIOENW5cdi49MZtE-4vfbSWO6dwKz0yVmbW0Rupct0-Fko0X8vZ5Cs6FQynnsO3BO9ywMFurK3GT4G4lAyfnWFvjs-RMVb5OXqn35UOzVdjXKS5coGKl2VfOv5r7N7b9IPpUeYxEKn8KiggZMYWPrdVgPqidwPQhvHJz0uF39AhH43HKJ6g01teAKmAiDSt_rxJVQNTwwcZhCpzFS4tJUg4BcqZ2BRf8QsPVDjhAi937UgYQdULJzQDBJ-6-brWGrWjZz4b3EfsXV0lT9Ynb5m91yHM6rDfexW7qOrt_VcosLmsbcbBmZ8UapjDCSyNTeHTdDmkBLxQmu2EpkYSqTjpt1cBaOHlb0Kj9NEkm-ls4rliZMw1WZx4HfhLxEBPWdGrMbpa_Ht1E35J9Gwe8Kt_B0p-R65zp0CmVsyRBOLVUZuVjiOF5uZqBWSklDVPQs-aF-4SweEF6XtORB5ha-mGq9UUh0TChXYKATxdu6OwwkL7NP_qgquoAm9oruyFvBpUornoI3WMTXKlVh0TrMloGHPVUUYGlGBjWcgPOf67uUBamvNpJD-5mjWo4HeHUy0UV1Bjc7vh3u0s=w1572-h1132-no)
-
最後にLogAttributeプロセッサを追加する
PROPERTIESタブは既定のままで、SETTINGSタブでsuccessをチェックする
![enter image description here](https://lh3.googleusercontent.com/qdHr3m6dlP9LFa5bn64CwR9e20QrTPKM1LiMlDBCB8XxU1VFZ6C2wnAbdQALrFv8vjha2PsiUD6ZXMTt_pdbF75WtL0yQ_JX60BlSRxr9n-mm9VAjDPgyFyxkk_mOxJr7YjwrAbCiBWfXyd9QZoB34lRHic7_I8F7I4YQoTMkEgQelybHVppD8SwaD7iqTcJfrOP5VKY4y6nqszaTTAAnXAbPsCrNrTnOOhrMuYIQryY4873a_vwQPRAVdPkRe7-9laQ05KqvZJ192BA3P18DSf8QYiECH7Qn2LY3P4GEGoyiBSgBo2D4JWkTzxNFVWo6OrDZHBUD5h428F2o769VUF-TUzhln5WMm6UF-eGuKFTm-l2Mc_7MLRVjVQOQvWRNkO4itZDe912_JYWl2XFqPRezRxh8OiS-zJWZiku-PjMEO3TOHlyozkN7-gqIQvTWxiW0KQ1AUCcXu90I-3k2VRYMlBTZoErDtUzdp2dP_GY7PvogKiCxcKm2aYloeny8KQtblh5SiqClLo_Grf8iJNlM2kOygLMU7etWoghg4mSOiMrmPMdhY2IuA0_WmdIaFtZL0FNRw0Dr1RlXt6vdqznsSvLfz8P6ZEGhqF_RP0GYjq8Eq-EnCdRzz7dXsOKGKpe9YEgXuPLo5qNvSdIU1mIr18g9Bg=w1574-h1128-no)
MergeContentのoriginal, failure, PutHDFSのfailureと繋ぐ
-
HDFS上のファイルの中身を見てみる
![enter image description here](https://lh3.googleusercontent.com/VSWrjwPmfgcMXVXNeKz1l9Uk_0QQqgdh9ggLKWu83TWzlc1EzSkIbGn5_t2b_w9F6xxZprbekld6-H4lodfm4k2tQo0utUEa5ApJJzhvxaIfOCx--ydkwQPHjY1TozKnM_T_NBVOipxpTfYSWFm9vTO3xC_wmNdA7vmSlfJfwhmAA5oRzP0I7M8A5fZh50HdsclvmII90vS3TzT9W3jJO32N9j5NlnTtD13PoC6L_C-lYhxKOtVYVUbchcZfZyCjaAnsIg81FY_AomebWugY9zlb7YtOQ5JzIr3fJa-flYVXbbAV60_GAnkPTvEdWBpyGi8EbDsIalsgQjc1FqwZs8JTKk6iLxTnoPQLCy6N7ZTd0XFXZWCxvMSVOJvmQMRM5UL_I9DIoYxUp_7hnc3bvcx5swobttDZBYT6MC0rRM4t8RXXl6CnR8zwZe9MEf_ofzIGMR0tFl7K0XViLYwNssf5OMOE2lfdGCwGkEQWOjPfzW-A3FjxrO08FR4NgdKqUInhD_FM9HOU9_0XiMIMJQQiloVjlHPKncdzRg7G133rkLYzrS2SNm4Pt9BWl7A0nIxPmpPFxAWga3HOdbdm7WkDpYPjgNiQncTri-zYyf_fS76n50MimN-qLTpoZJc5xZMjZXmKPSdx2Fmmva1hm_1mT0ANBHw=w1472-h532-no)
5、NiFiでセンサーデータをHiveテーブルに保存する
ここでHive Streamingを使ってセンサーデータをリアルタイムにHiveテーブルに追加します。
これを実現するには、Hiveテーブルがいくつか要件を満たす必要があります。
詳細はこちらをご参照くださいStreamingDataIngest-StreamingRequirements
1,ACIDサポートのため、hive-site.xmlに以下3つのパラメータを設定(HDP3.1ではすでに設定ずみ)
hive.txn.manager = org.apache.hadoop.hive.ql.lockmgr.DbTxnManager
hive.compactor.initiator.on = true
hive.compactor.worker.threads > 0
2,テーブル作成時に以下を含むことが必要
STORED AS ORC
tblproperties("transactional"="true")
CLUSTERED BY (cloumn-name) INTO <num> BUCKETS
下記の通りでテーブル(sensor_data_user1)を作成します。
CREATE TABLE sensor_data_user1(id int,time_str string, pressure double, temperature double, humidity double)
PARTITIONED BY(date_str string)
CLUSTERED BY (id) INTO 5 BUCKETS
STORED AS ORC
tblproperties("transactional" = "true");
この部分のデータフローは図の通りになります。
![enter image description here](https://lh3.googleusercontent.com/5g_Riel5ja0kFUTawn-PhaWcgQmpLLrE4L6kbRvJ5ODbHqHJG-lhu66ppHrYasQfrAZCKMke_AWxAQTsmPQH8eVIZOJs1HknzLB_OpowozWIV4Rnjdva-HpAbV70dfZf6_3UErPfbkKUnctIo11Admpy1V_pcEZREZ07902soIdvxHQVy27L6DY5puejygHuQ6INUj--GKecTD6Cx4HCso1K04TJqCc5IFV5F1K_-rdnaz3iWb3KCBbc5qzS6JcrfGk2SOrP-TSCDbOV8Ut3sHUD-g4Z7KduA6jkHgrlB3GnXhItq74f-kIs4vYza_OoSezQcJN6VncdYWL20xsUQNHaRc_2LsBmIhadU1xUdfvb9txgc678QYOGve3wWUX3l3yWkTz6zLul9VcRRza55ANCU84ZE5LQ2pPs0XNPrv9CXivJRd4RBBxubLNaNN-ELQySp8JSsHOoDPW94YpfXCEZIlDdG0H__gbg5SCib9xhIy0kxR6ChxkJm_O7F6-KP-xkz_45pG0jmu5aQhgWiFYyVipV7IVgnHtOaxfCMK5X8oYChPEZHx0fT5XiZqn7zLWeHgfXhhgoUiWow_Im64lHe8eB5qexbhw_dN_9IE7B6JlEuoDT1VENJIPlwWO2T8AaWet4HilSQjF2FbW3lShF8XnQv6Q=w1090-h1314-no)
これからプロセッサを説明していきます。
- UpdateAttributeプロセッサを追加し、FromMinifi input portと接続する
![enter image description here](https://lh3.googleusercontent.com/XbMcKN8OR7qXlnQaBLuwKC8TBWSscy6BjLjbR1C5Y1BiC5MrQiDjaNNkwTTkZbqr-tpEYehlTHiwJ2jLYhsjmO7cGBwvA2G9Ns1jH-5jM93HtJUAW_V6U1T7cY0xvBInsBjH71fi4es39GRhpddxClQbqBdJGPjV3Y3lig7sXCdpRoFZf3m5OROOHKiEpsJpKigHTcpxpkSUv-iCjKHHoDX6gkXrFMV9vBhKJuipE5SuMX0qiYprzkx-keNRZrB3ccoRExL-UThCxJRWKPoV2Wy9p6TFwe5mLMVMeD6ZOF0qRnsANf7qHo_zx15vfdjrt2bi4S-YUe3PY3Rmb5zDVuldvz-vlvrHvsqxh6U-CVIkq_A1o4M3Ylbg5E8mLi8VWdQkwDsAn1jAAMEw51lVIvSIhBMA5PyP5sy4SvnN30YpgEMlMjDNG9vulGYsfTFg1CLK10M5GL_ps3kEeXE55LcsCSGYDZLCEnqv9cQpcvuoult2xixQuz6Mk8ZH0edrRLtQzfvHEsl8WTak7mC5PBK_sjnVmRWdXiJllRcmX_g9wqFHqfgt6Ydb8A5TCwMHC13L-MWhVBzwk2sRkOGv_uF1rZ2rV_gQdI01O-OZeOR4r3ZtZP7DUy-wEycjuq9w42c4-etTGqFUQYu7gfrgFDv5A5C8Quo=w590-h400-no)
- UpdateAttributeのPROPERTIESタブでschema.name=sensor_data_schema1を追加する
![enter image description here](https://lh3.googleusercontent.com/AfsjX7nWnbDy2kyGrFLE7l1l6u90E-DBAhdqvfri1oE2vVgUmFrBF7LES0uszeHTXHLwsANZo5ePqxl8qY1m1i77cCf6R7YqH36lMZrVdIk6ZBiBL3vHJYqeMRDDLb5OD10JYAdcjf1wdC0IgMkp2SwFBv1nVmBPpWFF6ARyQVs5TdGD1dgTwrUMKPXBjToITkEsZ8zXlS8F0C37Cj_TMsKKNLAvfkQ-VCQCBaMS_jtI9hg2i6uN6tsHE7TsWYrbgsETnuGArpPH6LbN_QJHA9tcg-YyGtAb5vVujyGeFeXDgPdD6V6e99QsXOJgrLZsoZ8t9rPGfFbJC7f3RCm1gp3fy3bQRITbJhSDmxqBZBnrTllY_JsGW4p6AovfLMN0VK795XTrVOgQcmDFhQmHifOxdw_m7E2WzNbFYispof2GvDhcRBkuD7SLFDXW2DV_Lfyn-0FbazH4w9rtkisFF53KF35WKagLrweC5A05aInU2q6jl5MNOR1y4j5BRZlt20AzeocDqnH3RJ8PhaZ5i74H6k7UHZVg_V-rQLObYn4G60w3_SfmClnbr_Ke_ATXXEZc2AnaKNwzbC1JPXLmIZPCy4UTPMMIjo-VF1WURGz83BQfpQRiiaUu-eVQdiYOO7PuHo1_7i_cmBuiwhjlG-FiwEkV8c8=w1576-h1116-no)
- ConvertRecordプロセッサを追加する。CSVをAvroに変換
Record Reader: CSVReaderを選択
Record Writer: AvroRecordSetWriterを選択
![enter image description here](https://lh3.googleusercontent.com/OZH4tp7zJL68cCu4EMrvgDZtbCKP-GoBcyQ8t3Z-vOVcXybvAelZM3q51RS4l6C8nhyEHeCRlqhRRbXTJLiblk_iKHHn3fNu2WVJRDG_nPSbYBuYc22OhtzeT-5kBUbRWILOxhohJAYzV9NJiZTKTyWDEBJT5QkWA1Ejhpr9kdA_v7-X-NZWOi3AAFc6teOeaPsLX6HlSz--vlz9FZUFY2HGzQTDA1q8GGN5LRP2oNXy699yHCBZ6_T8LgHUPKV9yDgWE4p2a9RqhUpGfMuvj4rwvBs2AuHHjeIcptqzgkEEjgKEq6vU2bsPGo_2VjU7rq5SAruw88N4jOpZ6l_MGaJIfrLkppmfVtsSmKl9UTRC0eGzznnOIjRRrUoQ8t8n-Y_kOiN-KEwB8Wf0BN80M6UInn8PBVZMxn0dVgiisa900k6LqJXjH3blnDzOkVgNaVHZgD0B-dm4hemt30k7mLgoJ_OQxX7rZxiZYDEyCfG1xRXn-qsN495D2ZVwh2vczftGUodxXaq66i3f5J11IGHphQtKZ2FaqpYS1eCM5biYKS6L0sk6eSR8_3NZxpXFHriGYjqFnyAQtdmqYxW4K2RQh5QtCdYz7SMSq5K6o4YfIUTWzssyN0cag1wnXlx_bH2cYp-zYYSuWuAW83Di-osWaWyrE-8=w1576-h1112-no)
CSVReaderの→をクリックする。
Controller ServicesタブでAvroRecordSetWriterとCSVReaderが追加される。
![enter image description here](https://lh3.googleusercontent.com/P50QeHxsfutY4Z-6282MHDAamh4-GDHe94QtPlPjrqJVo3Yp2_Oym7HgtQL_R7Uzf7ofYKPP8jbm03qgOFt099IvbWVULoB8ba_AkG-7cg5pfRQQE7Pdf26Rs2Vud5u2m4vcTBMMalyf05c175u_nxa0e5axZuPeOMQ2iumxgmBL6Igqccwbeh9coT5___Vjqmu8rx66aS-6INSvHyISaRCts4OmXVec58PwZpCoC2WCSjVC3-PYw3n9NFlmfNZKcfHg0xAf2J7JKHwwkdLzpoFC7lIc6njX_SJc-gME1ZjW_BABNs2fe9H0dOPYTKc8OiO3EjuCnbMeRPhzVziMwXR1cO4p9ghKi4S8nSJeoUhJW163XEl2b9EILeepZ-dD20SeOF3cUygihkV3Eo9pUd65sSE60dg4Hiu60Q0xJdjk9yDVpgnC61474kBibQFZJvAaA5fTfb2xR78rILqDMXqQ5IqvfGQJ1KTuYRWRjJ2y69VT-oyfCpPU_eq4TO52-YHriYLmWWTcR-ptKGWddSs3myZRE1SzWy1h08NtqZSN_HeQ0SRq5Wtt_YrJFWfznzHDgzO99zndu8Zw1lFzsFMq3S9T7oKadzRsofxc1FVE-oQjyFPrYDBp5yl72SoWs9OTB_kQo098onghL8fJh5TGMuFUpC4=w2760-h528-no)
⚙をクリックする。
Schema Access Strategy: Use ‘Schema Name’ Propertyを選択
Schema Registryで Create new service… を選択する
![enter image description here](https://lh3.googleusercontent.com/1Q35Peln3yfsVB0IJabhcWBKybL-OZyWrKcu0-rZxqHXO6Ms1BubZv-g-Ubgzu6-_fUHBXcEUxK_mWOXv_GKzIcVSLklL2rFpbGkX8-nlahXUEtUeMcFdvFFZcLAO818eHNRnuoDlI2hd2XL-YEsT9xQo_M6ndyht_kMa-7EtFKRiomhqZ44E6WpZu4EgUW2JxfFqrY77JtDMPclSDQMmHc07GZYA3LRwVJwhQV5SBZqYOs2kUGAGdcND3ExetdtDehIqYAxUysAHMcyGMqZtyIGUKOK3fteXQcDst9gwFAuqicpHq9jXonKcqc3RM3y0tPug4hYkFqI2fBw2HDp_qdjKBsJkanMtEhOk2hNfSxoze47qSdx6vZM93syycB6QUj7GYCb03T6kaN_eJ5Zk_0AzfCL6v-HeEbwN7ymOsMY62rxZYcopBVGxH4XzUS4vPM23AKUPEthCRNVcTneYENMkN3rJY-ZsUIvPpTKZ5CEo1Gngjo5GcCrDboPVU5XtZBKuQhTh-ejcMEZou9I9MFIiP-0oghosQpIreiWE8p7svVDglepODzdkcH9T2nK_fSz5g7rQRr1G_SNurduajEhygvrQapyIlhj6t9gomhmtWLfHMYc6MSYd6nR9EoWBGKP1kZ-wU-riIxRNLVauBfiIVWfXL0=w1574-h1114-no)
AvroSchemaRegistry… を選択してCreateをクリックする
![enter image description here](https://lh3.googleusercontent.com/4VgierINRs0NDpjq5bE8BGPFAy8JYAAine-gNXkMQuz_2jzGwgHUBFDRWSaTJE-tdCmd4ZX2sV31LwVCrbmwtRFkE7cc79GD3k_8JlcLGKD1tnc1ZZzf6uwJHwP6JvQ3-UV2ecZGyEm0bU6BWpNi5v_hgVFbhq8e5_kycI0qed_vlmjQB1uBT1nQu1lbu7yHXcldlC2Q08dMDq8_RsKFxT02_OiXuBlOmtu6jipWLyKzVhtBK0dgBiJARJshHSV6lzcdjRkVQCtgxWBvoE2BzcB0-4XD7P48HEm43OEn_HYj_VdJVkInm3O-q58U2lPlb2WBHRC7Dy9WyA8ewup43dC71oyHTHxDKHHxuGKcNuLc6_v2xu21aoNM4_YtJ9Y8D2YAvVWTtgP1g6lQmBgOgNrJvCsiLwZSrl7_XyEWcFAAU1DLjONHl4V0ZqWy82opnTGaefPijt2Fp7KNxNqSGLfCKagLmYoUnfsBlQzfoFQ5k7MJqIZoa-0GtqfjUix9KGjVwhlEflerr_qsFOLSauw6Zq84xUbJlaM1qS2D7wnt-TjW-yhHa35EN9WYuT8t-rK-xdvZvQMkpv2cCLfvRCu3brYAzSLnlY6hpPbEjm-MDY8z4C50LJ8AsiAlOn1UVyRVBNH64jUrPZUOep0cUSJkNbtVRnM=w926-h836-no)
Schema Registryで AvroSchemaRegistryが表示される。→をクリックする。Save changes before going to Controller Service?が表示され、Yesクリックする
![enter image description here](https://lh3.googleusercontent.com/mpvpIbTnbbBjNgy3bDR8Kdcs8UHDA5-tSPK4Pm5QzmbLf4mF8-UAdQJJUnRhJFzuUQL-CUQVGzkiQ5vQ30FHlXK69J94BoW_eDsT_bA7Z03yyr_luh8IPTUbDscpsrRi07hSiSgz_3QMd4bR7HU9ZUyBQXNk6rEB-m75ewblj053l8Dw4SHBwYI1KfQjaNnf5tNvrzTSMCinjqaa6vIPpON43bBycriaFs-In1Df1wr08fwPRB7ZDLyGwCu8ZXyu4PLN9bYppTQuBCSU8iPzo4PWkeDWhU5p_k84lRpoCQUpRzSgbv4QT9AVvYNrK9yrGh1S7CzyAZ1xACKiZS6esWk1p03BEI70b2h_6J3JJBj8V5SeqLLlJMwOqHQWMm3FFMn8dw3FjlJc7_wtPF98rBZoKn5I_jGU16cABP2qM_tlTCtfycCmD5HB-mhUmeH7OP-Ebbqq4FKZxWX-mmwRIgv3pQLj0M3OPtDxt_ylZq6A8Jm7fGgtAPyMeopIMes4IKkh8OiFr3MrQzLLb7UEhCLg0-lbr1y2zaHaGApvDDzBk9Hy8b4msYlKOHXG22PhqFL_rzpTNpUs6V_YrJQDHVTRzcnKXrsnhh-_3EIEZk_4ZFmMe_wGHM2hiYlT0erXRhnRjWWqedz3cWWHQ00GrtUJgiT04Rs=w1572-h1122-no)
Controller ServicesタブでAvoSchemaRegistryが追加される
⚙をクリックする
![enter image description here](https://lh3.googleusercontent.com/YDBmPscpeN_bmSH1q6mrtuPiJEu9mQ8bCChJ0kMnCHs9hbuWbq12iCjRetk60UJ8cHq61jUI4XkhjOp_Z7UVQu6zMClVok6W_dmrskKN7nEujcQIa51-ihqWAm1E2fN2XVj8OFMbcHHKvKazfR5bhDUtA9Em6GNyYIMyaQwVSmuBxUKgP6KrhIZJGkp1CaAv6YSZWKTBTijp9tHwGdbXzA7TctTA18PPpalyNLAqsG8U6EKdD_CfOqlCNxhoUX857FpnG5tk51aI4pRyEo14lscKZO9i9-8KdHanH7IkFnkmt_9i1X6loz3m3aYb2ceZJUP-RoOu3i5iIRPDHTEQOYWA1s_kKsW1acEnQDOvnMuae0DqIIq1L1b37LxDrp-jBQoEsWhtkr_T-hjCXhLGh_mZ0Xt3MlUJ4KqtamvWVkol2Y5zKg4SyI9hovL5XBDF1ePyROGNGY-l_Qbdo8irn2UBBV9GWD45XTVKcU209xErpuI9_jQjUFY6He93368PiWsWXhJwo0nOHnUQxCwQbRod0ws6slfNSC0huPsHlbuTZ38xjnv6IrDUZrbV_j71ONKLX_mTucAar7acfrBYw9g6eAwnjoerXip_Ulzo0qD4vEV69zNOY-UGeKJyLzKc5dALcDLq2YBNjxj3rX3z9FTcpG4M0HU=w2746-h532-no)
Propertiesタブでプロパティ sensor_data_schema1を追加する
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110403/23799528-d62d-e2b6-6aea-7bf9a06b69bc.png)
Avroスキーマは下記の通り
{
"type": "record",
"namespace": "sensor_data_schema1",
"name": "sensor_data_schema1",
"fields": [
{ "name": "id", "type": "int" },
{ "name": "date_str", "type": "string" },
{ "name": "time_str", "type": "string" },
{ "name": "pressure", "type": "double" },
{ "name": "temperature", "type": "double" },
{ "name": "humidity", "type": "double" }
]
}
終わったら、CSVReaderがInvalid、AvroSchemaRegistryがDisabledの状態。
赤枠のアイコンをクリックしてAvroSchemaRegistryを有効にする(Enable→Closeをクリックする)。
CSVReaderも有効にする
![enter image description here](https://lh3.googleusercontent.com/avXEmOECbywZmeqObwpiPxrQNB46SSOkLAy3-4w-bGDJyj3KhrPeUiV8BoCHddLNdR6a32Kki-pa10l4ZhkygfzPGMYZz4TVUIW_t60GEeqEBtAnVr0KdBjQ5WBDRMslkhTVVDnqOwaWAFJl2Nce9oUhudEEx9ojcvEGLDDcyg74tdoINedBxz16TPXCaRh_U2KvECjTiwegrMH98ujSR7GFtYSoFnObZ6qJnaK9eRvColPVnHcZoRYIboWraIrWftcpy6prb3DDZVSpv4wOkGC4FTjvUn813VV5oQ9uZzp6-iHkXY937JEyA7tiOxNQ90x0VF0T-XNkTgakodWMeLHfq6C42M6Ejhp2FhroaFDILW_1CTTebeSRDjKbpNorDmsh8eU73gVe9OyHR8O1xMFSvhKkltxdnojzEQj5grWQLRsHMDULX_O4_svO-zlk6tDpfbw47o65KogFz72uNbj47qzH5o24x0UQo2M9YEN-S4nO-bE2PpZuUQKD8BnyLAcDnveUE6artpnJtqqWEcZR0IAUOv-pZam3N6KekOddU0nyigvHu8ZrzTANmgPg8NMRmLAbiQCFPQ1bKmT98P5Qo9uYTO_xqHFalTaI7AaSsIKNo9iHiS3bxmPwbXdqFizsDslFGi38WR4n3B-IXDSrOR-qeOo=w2754-h540-no)
最後にAvroRecordSetWriterも既定のままで有効にする
![enter image description here](https://lh3.googleusercontent.com/Bf665rUIHo1QYbjKJquh7ljgdwTBSg2yYZAkojiGBF8V4WroD8-LmLaNjzlyvCgb-zwlms0rViONGXX9aZThoI73GlGYw8aoEsqAMUVXwG1loOhLFBAhcDiVj3M_LWW2N9P3P4oAXRrDCGydu7R4wgSOWKNsu8imUYFjTuu4YWjtthTP-_iRcY1DGNH5KquO2NHDsiTlEdZczsuM2ytINswDGMjzxXmoJsOoAcyyjMgeP5KB4IMslCfIyLg7ULRY8BjqlTKy_VUycd7-OeRG8JYk0ct6lLJNhOLcGYkx_jqOcTXthYLKpXhG2rg3bwKZ-GoFvG1na27IBD5_VMqV2LqnKbkWlsaMdoVvzlN0JVj97n0iZmyyc326-zw5zYUaOeFWcMWAML8q_rfY1gLIl1wD4cnosnuHqWoN4nvYdeTljGjp1n5AZWHVx9qAvSemVr6xjtLDkb3Osal7FHfL-joDJ4_WIpcx4q4Q4mTvW81UclaEl9BeYoxAtHfQgsaN199CZOV5_kzGJgBpHZDkDqIsq7dZynp5546daS6pN-BGj0cOsdVexRyN9x2ugTX6sqyV63tJ-_XpVevrzhEDjctVh_IvxO5yNEZ0KIjjQtVCoteF_w-XZMg-vvwud3qNFHq1hqXUCJLcUVemzxq0bRKuGqqnfh8=w2742-h542-no)
ConvertRecordのfailureをLogAttributeに繋ぐ
- PutHive3Streamingプロセッサを追加する
![enter image description here](https://lh3.googleusercontent.com/90nuOovXcSqvWmdtt7FvkUG_cE4oLBjudyk1q8epkLE5SQcL_7uFl5VogjtZZY3HCvKH8DXCizt4MOB8YBDd4QnZELAyn8mykhoRj-QV6YXKHVtYKMtqq2Jp2OvF2n-3jjuq9Npehx2mdr85S7fkJ1UACbSOpMayujsVqAq4RbuWuaNxsbiMGLLUFAtbzCczfu0tPwbvNRaDiIfXwbmDK6j6zSSLjuiCOospeqVT0xsiYtAsHXhpWb6OI2rTRsQ-Xu1M8NB0WfYxTJ46XuwdQo-sw9D6uFX1oIDpQ_ea-T_GYD5Fc7zwuGE30sxOqK6Jn5dMo3raynA06Bzb2t_iDUcrFnH9dLre15ysjegreL1O6GrFjAIBhh8XC_qPQK24VgEiOFCU_PWBSeCDf5atWmwEUxVM4H_lT4qqv32O_1YPWHy2Ob91TsNRYPxizvcJicaowpcGO2ttLiV-3qJCuO5wcqER11LGOv_AdGy1dmnpfDXGuDSywMJS-w2sQoBaKbWhWSVDRgFtyChk7y1rf9RYYVvt2ODKcGV1Wh1oe3eUVncNoROHjtup4KX4MlhtMQ5NpcUxUrtplbKGqVpZgbaPC1ecW9lhdAqQFPGIzDIcnIgEJWD6ryR_NNp7Sh-BQsNzTpPRyGXvuBcEkTkLkMLiSPmb378=w1024-h882-no)
PutHive3StreamingのPropertiesタブで以下の値を設定する
Record Reader: AvroReaderを選択
Hive Metastore URL: thrift://hdp-srv3.demotest.com:9083を入力
Hive Configuration Resources: /etc/hive/conf/hive-site.xml,/etc/hadoop/conf/core-site.xml,/etc/hadoop/conf/hdfs-site.xmlを入力
Database Name: defaultを入力
Table Name: sensor_data_user1を入力
![enter image description here](https://lh3.googleusercontent.com/HGjqHaflt9dtv-Z8aPoY4WuX_r6MoRMiNx2prtp6EFPDeeVR2Fac6kAbC8xTZ9AbJHuPUmVa9wwLqa4IJ__pb1Czj1qvvkNdGvBad9Er4TfhAKJCvAgj5ILJwecvWQQFxeW57ZAUmqkjNhT5b2C83hoDkD3vFGnpjA5-HpnOZAK9CsZkfpmpCDD56K1wkb1gA5GRf-tZ3TESm44iXK7gpka93cO-8MQrWHvGjrggC8gb--gdlv9TU-y7uXL5edpQKyWLngEV2DtsksbTu8WQJrFa-WAkRPMi3dz3gq3AxcP961w9Yv8IcK2NMQBSIzwL1SG747ecSj3jjC2_BYOmYhUXTCnckaM-MbZiCR_TYszu53v4z3c-NbchLbmmOA_cJeijeCJCbIoy_qdYJEfcv_WVcdhE21l1Fte-a84QdyzePwTV349yAlEVYiH9rBoVGPLyRpG94eb4OjoTPKEk3D0L_4nv8h0ZFlLmq1-kqth_qYGyssW6-rT6YD9TMYOri4-ZwZ92s5DTbUeEhRnHwD3hDR8bYTk5hiuMCNgdjiHejed6lQsiLmtlSaQ_DdMR9KFXEP7dLgLyB-BVZhEAu13Pfjc54_qH6_i1sSlv3P5yFPt3FIdB9HU4PIcaDp9DiGfZTNZzTp4xEWzfGbCOOCBehP9Ryr8=w1572-h1112-no)
Record Readerの→をクリックし、AvroReaderを有効にする(Enable→Close)
![enter image description here](https://lh3.googleusercontent.com/iD9M5l8ewyqVmhVAqs_QZgjAhyUxGCGaSFZ1qGrI1Vt_RXWUoRD-3O1uHL7fAMOsGa1bORC_7NYGzxpuRlO7rRTta0bhjYobnhsM00nDJNEnyv97I2Nj6sayjzBF1JBbVVMM0gqz0Rgix9osKs4WrdxieipkMlK3cAnTg7yOzM7kpdYklws60OOa2zzct-Xo4o3x4hdoo4sYcwqOQLwhOOW6pGuu0ho5DaBRT_j19B06WCv9t6kZO4hZhiY5vRO83C4YdRtP4072dHXyMIO7ekLLL6NNMfGbU9rPWP4Jm5_QHnjbKUuaEnltBTVbKSiAOsWEMadeOilSAYVjuwk6rYnGGes1F0vBISn7TOiyQDBkCi7c2WcOy2StjwOR8nSX3k_fRS0yp8IbUPxWfe2B5NVwYI0gbJHAHji0F0nv3zs39BhEun8wn0iSAX4JVjRFVaxqBsEwUKOrfF6jJQF7uVSfPttJYGMtuUNPemCGvwijEaluailIoJv03ASVzxz8IRQjc6-iSpi0kSeTwDRJpJG1-vAYcTA41iIgh17jwo1LNZe4mfH7XF9Ov1YEx0hvm63-vgHp1SHiJWQ_2cB8BLumi05FH2XMc4RYDPypXkW1tOpIPvNg-GzoGKjS8SdFdwiY8MeWIBMtIFlcG9PonqAWP8iJMA0=w2742-h572-no)
PutHive3StreamingのSettingsタブでsuccessをチェックする
![enter image description here](https://lh3.googleusercontent.com/9VSFKI8mrb2uPiaDf0JBTciU7o8akV-qxxlzKFZFUhfo0tBR-CmStW8aorKe3TPcLx_WyJ-1xobWStd8J50XX4lYR7vOLm63ywgV2z_2uela0d3Wkju6Cat8K49RuaTCUMMPGSatBoZAKLSw9p7fcJBQt74rvi-5Yg58ClBqUmhadHS1cWZCskr0Acc-zSAbf_NgAoPoX7mIvGk0owFbIdMP_LZ8gBbxXifnQ3Rka6TEQompVaHLURkMFnV2JK1RZyTg1Bbx7FEs3vdv6Wq72iTSbkPLA4gcQT5LLRkBmaTnjLaRBKp8nE6osHGPkMuj458EFM9oekPu7ZxElYgButcv7qxXYRkDLfC_2TluSuJwjzI6XgJO9IWeulx612gn-tk3c6pOQG0tgUTs2KAdu5b5L0QcaFjtdRmm6SEsOnXvyCiN-RaSWcsKHiAdyTj9NfkkdK5vrJ2Sk_zyE0-UB1mGipLD7Chi1P2yTENPqueH02tuefBDIQLz71OouuGSAaKwYFOHxtQhQGN7UXjC4WPJiNuAAOz8sAt2JEfkE7k964z61YqH4Otm2rYvUqdp-YgaRoHYXLt8x-kSq01LDlXbqhxmwxaW8YNvZqJVprKFg7rJJKHlFHjR6Gf_7EEaRojTWNVSR3yFDDQL5O-sugv4KzoCzPY=w1580-h1118-no)
- PutHive3StreamingのfaiureとretryをLogAttributeに接続する
- Zeppelinでクエリを実行して確認
時間軸で温度の変化を表しています。
![enter image description here](https://lh3.googleusercontent.com/BcNXY8xeys7mV3G9n8Z1w2mGg8hE2XI4L-RFZRg8ZWwKkbTDpzzZBmtnZzvvDfWxV6degZgzuh4aBDUCRmUcvET3LFY5QxiXwSjyC1vQh_oCCOIxBtbFwsSu9ievA_meC-L9ICZE1p0jCLJOkwFXJr5Db3xuqPzSvkPCIF1EYzr4upH1nh9i1qsfWcI_e_XNsMtnQGw2pwvXkiC8I6Ni8YeTsMFOi9Q4GpWhTq5rUnnuT6uESE3U46UHLd-g-pjDuCZY0khNBW3AIfH8E3fhMoedK_m0k1swJOb8SO5fz9RGdR1WTvqi8muSKroyzfs99xNj_lcj7dFAvKy0AssGTivMFS5vDNpOm_ojenDml1olKJDjO1lgPtjA-po1MvT2UG-TbZ6SuE_1w5TtyKJ0wqYI-cCen3HMhzm_WLjnBtuovc-Vy4747uoGse5cvS9WQ_Zh4cHO_V1V4VApv9AOnaQPKGyr1HFMCAQ5vr5TH-ALPw2t5QSiciuBrdSx5G7fkEz-4PvejNf9C0LwrU8Q0cUcvQM4Vv0XMkJRX70kDk5AsvlqvPSIsLOKYPkUGzRVo9ndPHy_loCurS4XZHC2NJIQHKQc3E6adrjX2QlxsomDG0LsUkx-lSq9OAli5vEE6AqFUKg_VbTHO9qYESY_Ik4N4tsKc78=w2810-h770-no)
6、NiFiでデータを加工し、温度が閾値を超えたら、Slackアラート通知する
CSV形式のデータ「1,2019-06-04,23:59:51,1011.54,24.86,50.67」から温度を取得して、閾値(ここでは30度)を超えたらSlackにアラート通知します。
全体のデータフロー
![enter image description here](https://lh3.googleusercontent.com/NEzb40Obw3RFdpwVLFMLa1mwOLw3QxQ6k78B74CEnJIuk3jv84qqEZREvntulyJoemv6pDalcm9vC8GoNgKActLIcrBH7OGDddyuVXjwqSDGxAg0b3vcNse9_kcslzlBqOWM28Ggh9Hr4iiXMTe5eL4v3GsDeOnC78UzFW2ljI6YggMR3ntvmL5_jrRc8pHmGyxqdRB0NGnaCDPt5tWGD43kE0IL6g_T_bLwFO4z1xqYywhwvovsTCBWu5QrFGoEv452sP54boLh_Eq7HEcFZFF-R18Z1H-qiK0FUIs4C2gxB-9Vcifff2v9tOm3GBM4LBEq0uGEzJjHwlTcDoNrm7vG6Qn897iQWlcULK1L7nwhBScXta-OYRLlfTDG-0rNBziIu2VYPCK7xwPlPPCKpR27HQ4r_6OjwFVro18C8B3KSGZRFXu3zvLkQxIV7orXIJHRG_benxtVxSA8e5KWbX-gGrVljX6Fhbj5c6BWfDdGV0U8X1xo8wEuGO_uUA1fCGi6pINMk7svP5N-5Id38xhdmqd4NELiC0MWU2IgiEPAl3wh9dQicKqeRuVDCgG2JNMUPyNGTL_mUKpdFTj-Wm2NUa4Fcw01-AHaz5YqFkQS6b_U2_7SFKfpPoOhLSikD_qyY6fQvr9vwVYo8_qzF1vDlN3ssOk=w538-h1324-no)
- ExtractTextプロセッサを追加する
Propertiesタブでfieldプロパティを追加、値を(.*),(.*),(.*),(.*),(.*),(.*)に設定
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110403/b29d69a5-1fcd-b5c8-297a-b7b75f0b68f7.png)
Settingsタブでunmatchedをチェックする
![enter image description here](https://lh3.googleusercontent.com/Zl3bf7Xg6io-H1yefg3hvodZxvayge-pPitGZ4E_uZIEzDl7S8GrIi3Tmfh4Xv-F7uigQKBLKCZr-kq4taqdjzPab7PxPlFJam35WOwJDEcYJhY03sikNj4tz8MBw0nR4UVNaUeN-i_aiMyXQ5GxohfjoIhR1eJmkRnDGCbk0iywIMRxVPpbCNYbGX9vK77aIfVjLtczzM4prBHDC_EvSX01GkEJHCMAv05ypSEKGReeA7Pp7X5-oQ_OUWaSKwKgEtYqhVIGq0G3mEPZxPsu22m0FWRLxOU-unJAqeko9UYaeERrpvMNvILqLrOoEo5m57Z9aOEmnGsCim-cOqKsnBed3d6Rp5xCzYh-zXHfJegJDx_ymqYNHSeBuNi6atv_6v6Dy0JuyxyhhC9FpEkWVAYQBSZzspUzndE8LoYDE0Gp1MsXo3IS84k8bAa0nmuJLsAy7PKFQefGNCL8lHag6c-ENaBdth5ylLU_hUqVWxyuiwy130PafubOy8FL61xLAHFC5XCJ4a3sfQJc0QvekM7OEzkvGt9sq8Prp0LckXQ1MbhDQDUhWO1jOhKLIq_H7gMBIMQGr4AdXFXWxsQW6jvUwOfTyuEDtXXNf9UFZvk0tLXshHGEBgtqMY9u3R2TZouCQFNr7Z9tJxJAdd_kWcG7_bxDh2U=w1574-h1120-no)
- RouteOnAttributeプロセッサを追加する
Propertiesタブでhighプロパティを追加、値を${field.5:gt(40)} に設定する
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/110403/550ff115-eb7f-e9e9-9873-8d2d18f16643.png)
Settingsタブでunmatchedをチェックする
![enter image description here](https://lh3.googleusercontent.com/LXgCNl1dbnlIZgavlatO07_Yy4U3O9Ru3lBV96vGS1hsSGnGsSnH3XkVSGKKGIuimTtleRtNYhUhSrcakBLV1cxRWZFnO3JX0rDrbTSgI-tuJBPUIwrMIZTPQq-ZuJcT8XWAVKqblCz4btIArv4_Ud-UsfUDvfy_SIzfH_wtpEkO-TlLSkISkdwSlqx58wruwgKJsRPFIPX-l1g7XmT3zkaxrGumuxfkYWRHar7nr45Obvbmtgy8zk07cdV5NqbPxtfd8h4t5yLBrbsNwdDVwmgJvSlIcRBYVTkM9jz1hzbTrxuVonaCA0nryxj9fyWFcsJ9tnAX8-sgHIDlsIecBqThBH-hOHXFQiM946fD6AHvCeMkDslIdmRYKwOUajURA7FKZGmSXoB0FeZYc89ZvVTNDkabgAM1sRerm7hbH3sW9GBu5hkwo7Dq1Tr5kdjU33-0X5nXNEquOAKSu6SkRKREyXd41c2ZgyXfNBRh_HwfzMCijzvc4Zip244UOcq1t-LhTHXoWt9orBqXKWCgkxh7D3gGuOtPlwmBwEA2jYtm0pXCxTLfBc402qUjMLFf6xpUp6ZaUWemdy2Vx_4y0oH2CppTT07Zo7oTJMGQR59dUO52aJ3Gwz4VOMHjkuYaxUwbkaSTNyVcE5I5hATKmiFdzDt-FKk=w1572-h1120-no)
Settingsタブでsuccessをチェックする
![enter image description here](https://lh3.googleusercontent.com/uBhyVr36kRRbZOTz_pYpImzLYwaFqAkGa4v0gBFeWm7W1x_YQLcdOlbfZjkTUYLOJJy8rcdDAyyWEWuGv82gdn2BLobaGupUpiE0qGz_1-Q3B74YgGOD_tuOrTQ4qQ1Y6F87IfJNtkkMh2V1iM2z12tnIXDb0V7ofaXjfjq7pLvRzroVOH7Z3im_rOJcpwsI_TY3pCIyyOhC25lDpA_z8soLWODKDiFw1c3-Lq5VZOKkV7fzLiiVFih4_m0HinM8v5Pbc6w0PvXa5hzeM4wSEqEXhFEaEVkaR2q7JkMYQ0voVpQw1jJTdNYBhz3_HT7UCbJh9qwgEOS6pFD4F88PUdvVeMUq-O9pcH33cUmYj0tTR9X1NielyR3nSfkrsXzMLMGibzKNGvhjv63NNHZ8N7C3tAvBhvTtMDPoKy1O1s_TSS9QVIRYS6HFIfB4yr8TC8okP4CryGdq_d60iOVJNOxxHD2coERLrUjHup3HLoKKr6yvU9pV338gKZsd_Z2PPAQ75EA6MnoXsgJ_B0Y1SjudseVwO7-33pCcnEN2ffTamukLU7i5DXQYBSGT_7bG1iLX9Pvf1-0haoUrAcsmmhun-cQbHQbHFDBzc-VwMC4GoiKyaw4Un-YK042Yj85MnVds3MD6IkP_3tsZZ88GOHG5bWSHFYA=w1580-h1118-no)
- PutSlackのfailureをLogAttributeに接続する
- Slackメッセージ確認
![enter image description here](https://lh3.googleusercontent.com/QPxGFR0-lf0sOfhmBm6Qc-bCqihLmCYXyCqW0Z3FvMjWOiGzOemsiuXmlhl1VLQYpt7gSp2eE-rRMG2XrowiKbTs5CqjcJLqq3s05JLgOJGhZPWdIr8GcnayRPMLq2KDcIN5AkO-hgXz9Ch12WdSKVqc4-_r68gdKb-PoMbbP1e7BsIwKGia-GHraTmCU23DxySyGgPxEm-YgxojuQTEUqK68xy_At1pKM0_hGrCOTWnJ7lSdxmFI7sDMmarqF5VO7LIuI0H3V5XrZLAgYIt2Xj7rEdS8bx0fZ9ZxSopeSO1rcMUgCoDNaPGktuioKRxSJdD05zX5zeEqc7CzXD1J-Vb_kEu99Ll2CuGCxkhAwd69N8PsHVBaF8X4U0TlnyxLRfZiDcdggMfDHiZx0yG8Ny4joS_SyElnXKGwsGb7dqCBHYaXXlLYykeqzIXVtA0R--_ApW6yp8dtjWTbO-5x8iZgCy4aOJhLvehluNW4rRW8w8GLSjAO7lVL5xlbE14rjL_tJirQyBMOLqamp2oDJYHiIEpJQQdtcJfyojQBIzxpPxdTGPqbSsqeNejClghGdnNHZTcNUM5d80OQDHtWgWvrf1uYMjlyqAFyHPYfOoQ8KyE8s3lK_SiCKo-pb6a5QM2UPZgbt2tH9pFCcwmVW_6SkXFLyo=w858-h530-no)
最後のメモ: 事前準備
# Kafka topic作成
cd /usr/hdp/current/kafka-broker
./kafka-topics.sh --create --zookeeper hdp-srv1.demotest.com:218,hdp-srv2.demotest.com:2181,hdp-srv3.demotest.com:2181 --replication-factor 3 --partitions 1 --topic sensor_data_user1
# HDFS folder作成
sudo su - hdfs
hdfs dfs -mkdir -p /tmp/sensor_data/user1
hdfs dfs -chmod -R 777 /tmp/sensor_data/user1
# sensor_data_user1テーブルにNiFiユーザーにrwx権限追加
hdfs dfs -setfacl -m -R user:nifi:rwx /warehouse/tablespace/managed/hive/sensor_data_user1