使用Apache NiFi和Apache Kafka进行实时库存处理

朱雀 2022-10-02 12:00 211阅读 0赞

## **使用Apache NiFi和Apache Kafka实现从REST到Hive的流式使用案例** ##

### **第1部分** ###

使用Apache Kafka 2.0和Apache NiFi 1.8，有许多新功能和新功能即将推出。是时候对它们进行测试了。

![93293-streamingwithkafka2018flow.jpg][]

因此，为了规划我们将要做的事情，我有一个高级架构图。我们将提取许多来源，包括REST提要，社交资源，消息，图像，文档和关系数据。

我们将使用NiFi进行摄取，然后对其进行过滤，处理并将其细分为Kafka主题。Kafka数据将采用Apache Avro格式，并在Hortonworks Schema Registry中指定模式。Spark和NiFi将进行额外的事件处理以及机器学习和深度学习。这将存储在Druid中，用于实时分析和摘要。Hive，HDFS和S3将存储数据以进行永久存储。我们将使用Superset和Spark SQL + Zeppelin执行仪表板。

我们还将通过Kafka和NiFi向用户推送清理和汇总的数据。我们将推送到Dockerized应用程序，消息监听器，Web客户端，Slack通道和电子邮件列表。

为了在我们的企业中发挥作用，我们将通过Apache Ranger，Apache Atlas和Apache NiFi进行完全授权，身份验证，审计，数据加密和数据沿袭。NiFi Registry和GitHub将用于源代码控制。

我们将通过Apache Ambari提供管理功能。

**示例服务器布局：**

![93296-streamingwithkafka2018nodes.jpg][]

**NiFi流量**

![93282-stocksreadrest.png][]

IEX提供实时免费库存数据，无需许可证密钥。数据流速度非常快，幸好Apache NiFi和Kafka没有问题。

![93277-stocksconsumenifioverviewflow.png][]

从主题中获取不同的记录并在单独的目录和表中存储到HDFS。

![93289-stocksconsumeandstore.png][]

![93291-stockspushcryptotokafka.png][]

![93279-stocksconsumerecords.png][]

让我们将一个大的REST文件拆分成感兴趣的单个记录。我们的REST源包含引号，图表和新闻数组。

![93287-stockssplitjson3.png][]

![93288-stockssplitjson2.png][]

![93290-stockssplitjson.png][]

**让我们把一些消息推到Slack**

![93281-stockstoslack.png][]

我们可以轻松地使用Apache NiFi中的多个主题。

![93294-kafkaconsume.png][]

由于我们有模式，因此在运动时查询数据很容易

![93278-stocksrecordquery.png][]

我们为每个Kafka主题创建模式。

![93285-stocksquoteschema.png][]

![93297-stockchartsschema.png][]

我们可以监控在Ambari中通过Kafka传递的所有这些消息（以及在Hortonworks SMM中更详细的信息）。

![93295-kafkatopics.png][]

我读入数据然后可以将它推送到Kafka 1.0和2.0经纪人。

![93284-stockspublishkafka.png][]

一旦发送数据，NiFi就会让我们知道。

![93283-stockspubprovenance.png][]

### **使用的项目** ###

*  **Apache Kafka**
 *  **Apache NiFi**
 *  阿帕奇德鲁伊
 *  Kafka上的Apache Hive
 *  Apache Hive on Druid
 *  JDBC上的Apache Hive
 *  **Apache Zeppelin**
 *  NLP - Apache OpenNLP和Stanford CoreNLP
 *  **Horotnworks Schema Registry**
 *  **NiFi注册表**
 *  **Apache Ambari**
 *  **日志搜索**
 *  Hortonworks SMM
 *  Hortonworks数据平面服务（DPS）

### **来源** ###

**休息**

### **水槽** ###

*  **Apache Hadoop HDFS**
 *  **Apache Kafka**
 *  **Apache Hive**
 *  **松弛**
 *  S3
 *  阿帕奇德鲁伊
 *  Apache HBase

### **话题** ###

*  iextradingnews
 *  iextradingquote
 *  iextradingchart
 *  个股
 *  网络

### **HDFS目录** ###

hdfs dfs -mkdir  -p / iextradingnews

hdfs dfs -mkdir  -p / iextradingquote

hdfs dfs -mkdir  -p / iextradingchart

hdfs dfs -mkdir  -p / stocks

hdfs dfs -mkdir  -p / cyber

hdfs dfs -chmod  -R  777 /

### **PutHDFS** ###

*  /$\{kafka.topic\}
 *  /iextradingchart/859496561256574.orc
 *  /iextradingnews/855935960267509.orc
 *  /iextradingquote/859143934804532.orc

### **蜂巢表** ###

CREATE EXTERNAL TABLE如果不是 EXISTS iextradingchart（`date` STRING，打开DOUBLE，高DOUBLE，低DOUBLE，关闭DOUBLE，音量INT，unadjustedVolume INT，更改DOUBLE，changePercent DOUBLE，vwap DOUBLE，标签STRING，changeOverTime INT）

STORED AS ORC

LOCATION '/ iextradingchart' ;

CREATE EXTERNAL TABLE如果不是 EXISTS iextradingquote（符号STRING，companyName STRING，primaryExchange STRING，扇区STRING，calculationPrice STRING，打开DOUBLE，openTime BIGINT，关闭DOUBLE，closeTime BIGINT，高DOUBLE，低DOUBLE，latestPrice DOUBLE，latestSource STRING，latestTime STRING， latestUpdate BIGINT，latestVolume INT，iexRealtimePrice DOUBLE，iexRealtimeSize INT，iexLastUpdated BIGINT，delayedPrice DOUBLE，delayedPriceTime BIGINT，extendedPrice DOUBLE，extendedChange DOUBLE，extendedChangePercent DOUBLE，extendedPriceTime BIGINT，previousClose DOUBLE，change DOUBLE，changePercent DOUBLE，iexMarketPercent DOUBLE，iexVolume INT，avgTotalVolume INT，iexBidPrice INT，iexBidSize INT，iexAskPrice INT，iexAskSize INT，marketCap INT，peRatio 双倍，第52周高双，第52周低DOUBLE，ytdChange DOUBLE）

STORED AS ORC

LOCATION '/ iextradingquote' ;

创建外部表如果不存在 iextradingnews（`datetime` STRING，标题STRING，源STRING，url STRING，汇总STRING，相关STRING，图像STRING）

STORED AS ORC

LOCATION '/ iextradingnews' ;

### **架构** ###

{ “type”：“record”，“name”：“iextradingchart”，“fields”：[{   “name”：“date”，   “type”：[   “string”，   “null”   ]}，{   “name”：“open”，   “type”：[   “double”，   “null”   ]}，{   “name”：“high”，   “type”：[   “double”，  “null”   ]}，{   “name”：“low”，   “type”：[   “double”，   “null”   ]}，{   “name”：“close”，   “type”：[   “double”，   “null”   ]}，{   “name”：“volume”，   “type”：[   “int”，   “null”   ]}，{   “name”：“unadjustedVolume”，   “type”：[   “int”，   “null”   ]}，{   “name”：“更改“ ，   ”类型“：[   ”double“，   ”null“   ]}，{  “name”：“changePercent”，   “type”：[   “double”，   “null”   ]}，{   “name”：“vwap”，   “type”：[   “double”，   “null”   ]}，{   “name “：”label“，   ”type“：[   ”string“，   ”null“   ]}，{   ”name“：”changeOverTime“，   ”type“：[   ”int“，  “null”   ]}]} { “type”：“record”，“name”：“iextradingquote”，“fields”：[{   “name”：“symbol”，   “type”：[   “string”，   “null”   ]，   “doc”：“从'\”HDP推断的类型\“'”   } ，{   “name”：“companyName”，   “type”：[   “string”，   “null”   ]，   “doc”：“从'\”推断的类型Hortonworks Inc. \“'”   }，{   “name”：“primaryExchange”，   “type”：[   “string”，   “null”  ]，   “doc”：“从'\”推断的类型“纳斯达克全球选择\”'“   }，{   ”名称“：”扇区“，   ”类型“：[   ”字符串“，   ”空“   ]，   ”doc“：”类型从'\'Technology \“'”   }，{   “name”：“calculationPrice”，   “type”：[   “string”，   “null”   ]，   “doc”：“从'\”中推断类型关闭\“ ““   }，{   ”name“：”open“，   ”type“：[   ”double“，  “null”   ]，   “doc”：“从'16。''推断类型'   }}，{   ”name“：”openTime“，   ”type“：[   ”long“，   ”null“   ]，   ”doc“：”从'1542033000568'“   }，{   ”name“：”close“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“：”从'15 .76'推断的类型'   }}，  { “name”：“closeTime”，   “type”：[   “long”，  “null”   ]，   “doc”：“从'1542056400520'推断的类型'   }}，{   ”name“：”high“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“：”从'16 .37'“   }，{   ”name“：”low“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“：”从'15。''推断的类型'   }}，{   “name“ ：”latestPrice“，   ”type“：[   ”double“，  “null”   ]，   “doc”：“从'15。76'推断的类型”   }，{   “name”：“latestSource”，   “type”：[   “string”，   “null”   ]，   “doc”：“从'\“关闭\”'“   }，{   ”name“：”latestTime“，   ”type“：[   ”string“，   ”null“   ]，   ”doc“：”从'\“推断的类型”2018年11月12日“'“   }，{   ”name“：”latestUpdate“，   ”type“：[   ”long“，   “null”   ]，   “doc”：“从'1542056400520'推断的类型'   }}，{   ”name“：”latestVolume“，   ”type“：[   ”int“，   ”null“   ]，   ”doc“：”类型推断来自'4012339'“   }，{   ”name“：”iexRealtimePrice“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“：”类型从'15 .74'推断“  }，{   “name”：“iexRealtimeSize”，   “type”：[  “int”，   “null”   ]，   “doc”：“从'43'推断的类型”   }，{   “name”：“iexLastUpdated”，   “type”：[   “long”，   “null”   ]，   “doc”：“类型推断自'1542056397411'”   }，{   “name”：“delayedPrice”，   “type”：[   “double”，   “null”   ]，   “doc”：“类型推断自'15.76'“   }，{   ”name“：”delayedPriceTime“，   ”type“：[   “long”，   “null”   ]，   “doc”：“从'1542056400520'推断类型'   }}，{   ”name“：”extendedPrice“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“ “：”类型从'15 .85'“   } 推断，{   ”name“：”extendedChange“，   ”type“：[   ”double“，   ”null“   ]，   ”doc“：”类型从'0.09'“   }推断，{   ”name“：”extendedChangePercent“，  “type”：[   “double”，   “null”   ]，   “doc”：“从'0.00571'推断的类型'   }}，{   ”name“：”extendedPriceTime“，   ”type“：[   ”long“，   ”null“   ] ，   “doc”：“类型推断自'1542059622726'”   }，{   “name”：“previousClose”，   “type”：[   “double”，   “null”   ]，   “doc“ ：”从'16 .24'推断的类型“   }}，{   ”name“：”更改“，   “type”：[   “double”，   “null”   ]，   “doc”：“从'-0.48'”推断类型   }，{   “name”：“changePercent”，   “type”：[   “double”，   “null” “   ]，   ”doc“：”类型从'-0.02956'“   } 推断，{   ”name“：”iexMarketPercent“，   ”type“：[   ”double“，   ”null“   ]，  “doc”：“类型从'0.03258'推断'   }}，{   ”name“：“iexVolume”，   “type”：[   “int”，   “null”   ]，   “doc”：“从'130722推断的类型'”   }，{   “name”：“avgTotalVolume”，   “type”：[   “int”，   “null”   ]，   “doc”：“从'2042809'推断的类型'”   }，{   “name”：“iexBidPrice”，   “type”：[   “int”，   “null”  ]，   “doc”：“从'0'推断的类型”   }，{   “name”：“iexBidSize”，   “type”：[   “int”，   “null”   ]，   “doc”：“从'0'推断的类型”   }，{   “name”：“iexAskPrice”，   “type”：[   “int”，   “null”   ]，   “doc”：“从'0'推断的类型”   }，{   “name”：“iexAskSize”，   “type”：[   “int”，   “null”   ]，  “doc”：“从'0'推断的类型”   }，{   “name”：“marketCap”，   “type”：[   “int”，   “null”   ]，   “doc”：“从'1317308142'推断的类型'”   }，{   “name”：“peRatio”，   “type”：[   “double”，   “null”   ]，   “doc”：“类型从'-7.43'”   } 推断，{   “name”：“week52High”，   “type”：[   “double”，   “null”   ]，   “doc”：“类型从'26 .22'推断   }”，{   “name”：“week52Low”，   “type”：[   “double”，   “null”   ]，   “doc”：“从'15。''推断类型'   }}，{   ”name“：”ytdChange“，   ”type“：[   ”double“，   ”null“   ]，   “doc”：“类型从'-0.25696247383444343'”   }}}} {} type“：”record“，”name“：”iextradingchart“，”fields“：[{ “name”：“date”，“type”：[ “string”，“null” ]}，{ “name”：“open”，“type”：[ “double”，“null” ]}，{ “name”：“high”，“type”：[ “double”，“null” “ ]}，{ ”name“：”low“，”type“：[ ”double“，”null“ ]}，{ ”name“：”close“，”type“：[ ”double“，“null” ]}，{ “name”：“volume”，“type”：[ “int”，“null” ]}，{ “name”：“unadjustedVolume”，“type”：[ “int”，“null” ]}，{ “name”：“change”，“type”：[ “double”，“null” ]}，{ “name”：“changePercent”，“type”：[ “double”，“null” ]}，{ “name”：“vwap”，“type”：[ “double”，“null” ]}，{ “name”：“label”，“type”：[ “string”，“null” ]}，{ “name”：“changeOverTime”，“type”：[ “int”，“null” ]}]}

### **给Slack的消息** ###

文件：$ \{'filename'\}

抵消：$ \{'kafka.offset'\}

分区：$ \{'kafka.partition'\}

主题：$ \{'kafka.topic'\}

UUID：$ \{'uuid'\}

记录数：$ \{'record.count'\}

文件大小：$ \{fileSize：divide（1024）\} K.

请参阅jsonpath.com

### **拆分** ###

$。\*报价

$。\*。图

$。\*新闻

### **数组为单** ###

$ \*。

**GETHTTP**

### **网址** ###

https://api.iextrading.com/1.0/stock/market/batch?symbols=hdp&types=quote,news,chart&range=1y&last=25000

### **文件名** ###

`marketbatch.hdp.${'hdp':append(${now():format('yyyymmddHHMMSS'):append(${md5}):append('.json')})} `  IEX免费提供的数据。查看IEX的使用条款。

IEX实时价格

### **查询** ###

SELECT * FROM FLOWFILE

WHERE latestPrice> week52Low

SELECT * FROM FLOWFILE

WHERE latestPrice <= week52Low

### **示例输出** ###

File：855957937589894

抵消：22460

分区：0

主题：iextradingquote

UUID：b2a8e797-2249-4689-9a78-4339ddb5ecb4

记录数：

文件大小：3K

### **使用Hive和Spark SQL在Apache Zeppelin中进行数据可视化** ###

![93276-stockszep1.png][]

在HDFS中创建Apache ORC文件之上的表很容易。

![93292-zeppelinnews1.png][]

![93275-stockszep2.png][]

**将某些消息推送到Slack**

[93293-streamingwithkafka2018flow.jpg]: /images/20220112/f881e6546eed45e39fcb4c956021318e.png
[93296-streamingwithkafka2018nodes.jpg]: /images/20220112/d2796a77415d4d99a9baf3741b5b14cc.png
[93282-stocksreadrest.png]: /images/20220112/5d19a641662f4758bba87207d5b96b35.png
[93277-stocksconsumenifioverviewflow.png]: /images/20220112/0216965d00e549dda381a68da6c9deb0.png
[93289-stocksconsumeandstore.png]: /images/20220112/ade6dae7d7ea4fcfb27152b79c68020a.png
[93291-stockspushcryptotokafka.png]: /images/20220112/744cca575497436fb6f39f81009c5c6d.png
[93279-stocksconsumerecords.png]: /images/20220112/4281800155b147389cb8255d06e6a38a.png
[93287-stockssplitjson3.png]: /images/20220112/811c544c77c741db832e3b76f805f0b3.png
[93288-stockssplitjson2.png]: /images/20220112/d0a1e321d0e6459083b2042388984cf3.png
[93290-stockssplitjson.png]: /images/20220112/65a8345a9a6b4247ba23ae40ccc693d1.png
[93281-stockstoslack.png]: /images/20220112/70ee9ed25ea0477dbe2db9e981279f39.png
[93294-kafkaconsume.png]: /images/20220112/4136e44ac9ac4d148be970ac7ea6d0cf.png
[93278-stocksrecordquery.png]: /images/20220112/1507a0744ccf488dbb313d4db161cf14.png
[93285-stocksquoteschema.png]: /images/20220112/c01e2f3525854f6a9f3649e27a1bb93d.png
[93297-stockchartsschema.png]: /images/20220112/6f47f94ebbf8436888b21665325f2274.png
[93295-kafkatopics.png]: /images/20220112/fae5f4b5ce8d489aa77f336d92889278.png
[93284-stockspublishkafka.png]: /images/20220112/87ec2484f6db49b7a1a13d8fef8e06f0.png
[93283-stockspubprovenance.png]: /images/20220112/8c3c9236e6e3475a8789d0660c635e2d.png
[93276-stockszep1.png]: /images/20220112/ca8191bc658849298035dd910b7bb517.png
[93292-zeppelinnews1.png]: /images/20220112/38d01be346f84433ae0da298fe3468cd.png
[93275-stockszep2.png]: /images/20220112/bfd26e33a5c5430893eec125cc781f97.png