在本练习中，您将学习如何使用 Amazon Kinesis 流式传输数据到 Elasticsearch 并进行分析，这是两项完全托管的基于云的服务，用于实时传输大型分布式数据流和查询，分析等。

为了使练习更加贴近实际业务场景，我们将模拟从 EC2 的应用程序中生成交易订单事件。

在此教程中，您将完成以下三个组件的实验：

• 创建Amazon Kinesis Data Stream
• 创建Amazon Kinesis Data Firehose
• 部署Amazon Elasticsearch

本实验的架构图如下：

构建宽表

因为 Elasticsearch 最多只支持 2 个表的 Join 查询，而我们的表有 4 个，所以为了更好的在 Elasticsearch 里面使用，我们通过脚本对数据进行了一定的处理，做成了宽表的模式。

准备表数据

登录之前部署的 EC2 客户端，上传这三个文件，放到一个专门的目录下，例如/home/ec2-user/csv下

https://imgs.wzlinux.com/aws/tbl_address.csv

https://imgs.wzlinux.com/aws/tbl_customer.csv

https://imgs.wzlinux.com/aws/tbl_product.csv

[ec2-user@ip-172-31-77-126 csv]$ lltotal 176-rw-rw-r-- 1 ec2-user ec2-user 91306 Oct 30 09:53 tbl_address.csv-rw-rw-r-- 1 ec2-user ec2-user 75857 Oct 30 09:53 tbl_customer.csv-rw-rw-r-- 1 ec2-user ec2-user  6552 Oct 30 09:53 tbl_product.csv

加载进入缓存

登录之前部署的 EC2 客户端，上传这三个文件，跟上一步的三个 csv 文件放到一个目录下

https://imgs.wzlinux.com/aws/rds_address.sh

https://imgs.wzlinux.com/aws/rds_customer.sh

https://imgs.wzlinux.com/aws/rds_product.sh

启动 redis 服务，安装 unix2dos 工具

sudo amazon-linux-extras install redis4.0 -ysudo systemctl enable redissudo systemctl start redissudo yum install dos2unix -y

执行如下脚本导入数据到 Redis（确保导入脚本和 csv 文件在同一个目录下）

[ec2-user@ip-172-31-77-126 csv]$ sh rds_address.sh unix2dos: converting file ./tbl_address.rds to DOS format ...All data transferred. Waiting for the last reply...Last reply received from server.errors: 0, replies: 3252[ec2-user@ip-172-31-77-126 csv]$ sh rds_customer.sh unix2dos: converting file ./tbl_customer.rds to DOS format ...All data transferred. Waiting for the last reply...Last reply received from server.errors: 0, replies: 4336[ec2-user@ip-172-31-77-126 csv]$ sh rds_product.sh unix2dos: converting file ./tbl_product.rds to DOS format ...All data transferred. Waiting for the last reply...Last reply received from server.errors: 0, replies: 500

生成流式数据宽表

登录之前部署的 EC2 客户端，上传这个可执行文件（例如放到根目录下），如果是其他区域，请修改一下脚本。

https://imgs.wzlinux.com/aws/lab4.sh

执行脚本开始灌数据

sh lab4.sh kds-lab4 2021-03-19 &

lab4.sh 会往 kds 流里面灌数据，格式为

tid: 交易id
tno：交易编号
tdate：交易日期
uno：客户编号
pno：产品编号
tnum：交易数量
uname：下单客户的姓名
umobile：下单客户的手机号
ano：下单配送地址编号
acity：下单配送城市
aname：下单配送地址
pclas：产品类型
pname：下单产品名称
pprice：单价
total：总价（=tnum * pprice）
tuptime：时间戳

如下仅供参考

{    "tid": "D123",    "tno": "DwGi20200904131249",    "tdate": "2021-03-19",    "uno": "U1030",    "pno": "P1002",    "tnum": 10,    "uname": "张三",    "umobile": "13800138000",    "ano": "A1001",    "acity": "深圳",    "aname": "福田区**地址",    "pclass": "MOBILE",    "pname": "OPPO-Reno4",    "pprice": 4999,    "total": 4999,    "tuptime": "2021-03-19T13:15:48Z"}

检查 lab4 开头的日志文件输出，看是否运行正常，出现如下所示截图表示正常表构建完毕后如下图所示

宽表构建完毕，可以开始在Elasticsearch里面开始接受数据并进行实时检索和可视化了。

实时检索和可视化

本实验演示通过 Kinesis Firehose 把流数据通过管道注入 Elasticsearch 实时检索和可视化的过程。

导入数据

登录并打开 Elasticsearch 控制台，确保 ES 集群状态是有效。

打开 Kinesis Firehose 控制台，选择创建传输流，设置传输流名字“lab4-kfh”，并选择“kds-lab4”为源

下一步不对记录进行任何处理（选默认值），在下一步的目标里面选择将数据送到 Elasticsearch，并选择我们部署的 ES 集群，以及数据的索引名（此处为 lab4），如下图所示

需要配置 S3 转存储一份（或保留发送失败的记录），配置对应桶和路径即可

接下来的缓冲区设置为 1M 或者 60 秒即可（其他全部默认）

然后选择审核并创建传送流即可，但是此时 KDF 的数据流还没有权限写入 ES，因为 ES 开启了精细访问控制，我们需要登陆到 Kibana 为 KDF 的 Role 加上对 ES 的访问权限。

登录 ES 集群添加权限

在我们这里，Kibana 地址为如下，输入我们创建集群时添加的 master 账户认证即可。

https://search-lab-es-lr5h4xkkutqsb4y6d4nvig5j2a.us-east-1.es.amazonaws.com/_plugin/kibana/

点击 all_access，进入 Mapped users

创建角色映射，我们把 KDF 的角色写入到 Backend roles 里面

点击映射保存

创建 Index

点击主菜单进入 Discover，选择 Create index pattern

系统已经发现了 Index 为 lab4 的相关数据，我们直接输入 lab4 即可（注意：如果此处看不到数据，请略微等待，如果长时间等待依然没有数据，请确认 Kinesis Firehose 的配置是否准确）

接下来选择tuptime为时间戳即可

通过选择对应字段，我们即可看到对应数据

按时间维度查看

接下来我们来创建第一个可视化图表（按时间维度查看销售额），首先创建一个视图

类型选择 Vertical Bar

类型选择数据源

在 Y-axis 界面，aggregation下拉框选择 sum，field 选择 total，并备注为总销售额

在 Buckets 界面，点击 add，选择 X-axis，Aggregation 选择 Date Histogram，Field 选择 tuptime，Minimum 选择默认值 Auto（因为我们输入数据时间较短，所以难以形成例如以天为单位的汇总数据）

点击更新后获得如下图表，选择左上角的保存

设置保存名称：time_revenue。注意：图表的时间线，可以根据要求在右上角位置进行修改。

按区域维度查看

这一次我们选择表格的方式（取销售额排名前10的城市，选择表格的形式）

配置指标数据为 total 列的和

再点击 add，创建 split rows，Sub-aggregation 选择 terms，fields 选择 acity.keyword，单击更新按钮

获得的 Top 10 城市销售金额表（注意：因为流数据一直在注入，所以数据会变），如下图所示

然后点击左上角的“save”按钮进行保存，输入图表的名称 time_city

按产品维度查看

我们选择用饼图的方式来按产品维度查看

配置指标数据为 total 列的和

结合产品分类

并且配置显示标签（如果未显示标签就只有图，选中了标签以后会显示对应的备注和数据）

获得如下图所示的饼图（显示Top销量的产品分类，以及其他的汇总），保存成：time_pclass

创建仪表盘

在上述章节的实验过程中，我们创建了 3 个可视化报表

这个章节，我们把刚才创建的图表做成仪表盘。点击左侧按钮，创建 Dashboard，点击“create new dashboard”，点击 add 按钮，把刚才的三个图表添加进来

添加已经创建好的各个图表（并保存仪表盘，此处为 MyLab4），效果如下

可以根据选择右上角不同的时间区间获得不同的实时统计结果，各位熟悉 Elasticsearch 的学员可以按照自己喜欢的方式在 ES 里面构建各种图表。

欢迎大家扫码关注，获取更多信息

喜欢我的文章，成为我的天使投资人吧

赞赏

0人进行了赞赏支持

AWS 大数据实战 Lab4 - 数据实时检索(五)