位置: IT常识 - 正文

Apache Druid 数据摄取---本地数据和kafka流式数据(apache druid使用)

编辑：rootadmin

Durid概述 Apache Druid是一个集时间序列数据库、数据仓库和全文检索系统特点于一体的分析性数据平台。本文将带你简单了解Druid的特性，使用场景，技术特点和架构。这将有助于你选型数据存储方案，深入了解Druid存储，深入了解时间序列存储等。 Apache Druid是一个高性能的实时分 ... Durid概述

推荐整理分享Apache Druid 数据摄取---本地数据和kafka流式数据(apache druid使用)，希望有所帮助，仅作参考，欢迎阅读内容。

文章相关热门搜索词:druid api,apache druid使用,apache druid github,apache的druid和阿里的druid,druid数据源,druid数据库,apache druid github,apache druid github,内容如对您有帮助，希望把文章链接给更多的朋友！

Apache Druid是一个集时间序列数据库、数据仓库和全文检索系统特点于一体的分析性数据平台。本文将带你简单了解Druid的特性，使用场景，技术特点和架构。这将有助于你选型数据存储方案，深入了解Druid存储，深入了解时间序列存储等。

Apache Druid是一个高性能的实时分析型数据库。

上篇文章，我们了解了Druid的加载方式，

咱么主要说两种，一种是加载本地数据，一种是通过kafka加载流式数据。

数据摄取4.1 加载本地文件

我们导入演示案例种的演示文件

4.1.1.1 数据选择

通过UI选择local disk

并选择Connect data

4.1.1.2 演示数据查看

演示数据在quickstart/tutorial目录下的wikiticker-2015-09-12-sampled.json.gz文件

4.1.1.3 选择数据源

因为我们是通过imply安装的，在Base directory输入绝对路径/usr/local/imply/imply-2021.05-1/dist/druid/quickstart/tutorial,File filter输入wikiticker-2015-09-12-sampled.json.gz,并选择apply应用配置，我们数据已经加载进来了

Base directory 和 File filter 分开是因为可能需要同时从多个文件中摄取数据。

4.1.1.4 加载数据

数据定位后，您可以点击"Next: Parse data"来进入下一步。

数据加载器将尝试自动为数据确定正确的解析器。在这种情况下，它将成功确定json。可以随意使用不同的解析器选项来预览Druid如何解析您的数据。

4.1.2 数据源规范配置4.1.2.1 设置时间列

json 选择器被选中后，点击 Next：Parse time 进入下一步来决定您的主时间列。

Druid的体系结构需要一个主时间列（内部存储为名为_time的列）。如果您的数据中没有时间戳，请选择固定值（Constant Value）。在我们的示例中，数据加载器将确定原始数据中的时间列是唯一可用作主时间列的候选者。

这里可以选择时间列，以及时间的显示方式

4.1.2.2 设置转换器

在这里可以新增虚拟列，将一个列的数据转换成另一个虚拟列，这里我们没有设置，直接跳过

4.1.2.3 设置过滤器

这里可以设置过滤器，对于某些数据可以不进行显示，这里我们也跳过

4.1.2.4 配置schema

在 Configure schema 步骤中，您可以配置将哪些维度和指标摄入到Druid中，这些正是数据在被Druid中摄取后出现的样子。由于我们的数据集非常小，关掉rollup、确认更改。

4.1.2.5 配置Partition

一旦对schema满意后，点击 Next 后进入 Partition 步骤，该步骤中可以调整数据如何划分为段文件的方式，因为我们数据量非常小，这里我们按照DAY进行分段

4.1.3 提交任务4.1.3.1 发布数据

点击完成 Tune 步骤，进入到 Publish 步，在这里我们可以给我们的数据源命名，这里我们就命名为druid-sampled，

点击下一步就可以查看我们的数据规范

这就是您构建的规范，为了查看更改将如何更新规范是可以随意返回之前的步骤中进行更改，同样，您也可以直接编辑规范，并在前面的步骤中看到它。

4.1.3.2 提交任务

对摄取规范感到满意后，请单击 Submit，然后将创建一个数据摄取任务。

您可以进入任务视图，重点关注新创建的任务。任务视图设置为自动刷新，请等待任务成功。

当一项任务成功完成时，意味着它建立了一个或多个段，这些段现在将由Data服务器接收。

4.1.3.3 查看数据源

从标题导航到 Datasources 视图，一旦看到绿色（完全可用）圆圈，就可以查询数据源。此时，您可以转到 Query 视图以对数据源运行SQL查询。

4.1.3.4 查询数据

可以转到查询页面进行数据查询，这里在sql窗口编写sql后点击运行就可以查询数据了

4.2 kafka加载流式数据4.2.1 安装Kafka

这里我们使用docker-compose的方式启动kafka

4.2.1.1 编辑资源清单vi docker-compose.ymlversion: '2'services: zookeeper: image: zookeeper container_name: zookeeper ports: - 2181:2181 kafka: image: wurstmeister/kafka ## 镜像 volumes: - /etc/localtime:/etc/localtime ## 挂载位置（kafka镜像和宿主机器之间时间保持一直） ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: 192.168.64.190 ## 修改:宿主机IP KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 ## 卡夫卡运行是基于zookeeper的 KAFKA_ADVERTISED_PORT: 9092 KAFKA_LOG_RETENTION_HOURS: 120 KAFKA_MESSAGE_MAX_BYTES: 10000000 KAFKA_REPLICA_FETCH_MAX_BYTES: 10000000 KAFKA_GROUP_MAX_SESSION_TIMEOUT_MS: 60000 KAFKA_NUM_PARTITIONS: 3 KAFKA_DELETE_RETENTION_MS: 10004.2.2.2 启动容器docker-compose up -ddocker-compose ps

4.2.3 验证kafka

启动kafka后需要验证kafka是否可用

4.2.3.1 登录容器

登录容器并进入指定目录

#进入容器docker exec -it kafka_kafka_1 bash#进入 /opt/kafka_2.13-2.7.0/bin/ 目录下cd /opt/kafka_2.13-2.7.0/bin/

4.2.3.2 发送消息

运行客户端发送消息，注意这里的连接地址需要写我们配置的宿主机地址

#运行kafka生产者发送消息./kafka-console-producer.sh --broker-list 192.168.64.173:9092 --topic test

发送的数据如下

{"datas":[{"channel":"","metric":"temperature","producer":"ijinus","sn":"IJA0101-00002245","time":"1543207156000","value":"80"}],"ver":"1.0"}

4.2.3.3 消费消息

运行消费者消费消息

./kafka-console-consumer.sh --bootstrap-server 192.168.64.173:9092 --topic test --from-beginning

Apache Druid 数据摄取---本地数据和kafka流式数据(apache druid使用)

有数据打印说明我们kafka安装是没有问题的

4.2.4 发送数据到kafka4.2.4.1 编写代码

编写代码发送消息到kafka中

@Componentpublic class KafkaSender { @Autowired private KafkaTemplate<String, String> kafkaTemplate; /** * 发送消息到kafka * * @param topic 主题 * @param message 内容体 */ public void sendMsg(String topic, String message) { kafkaTemplate.send(topic, message); }}@RestController@RequestMapping("/taxi")public class KafkaController { @Autowired private KafkaSender kafkaSender; @RequestMapping("/batchTask/{num}") public String batchAdd(@PathVariable("num") int num) { for (int i = 0; i < num; i++) { Message message = Utils.getRandomMessage(); kafkaSender.sendMsg("message", JSON.toJSONString(message)); } return "OK"; }}4.2.4.2 发送消息

使用postman 发送消息到kafka，消息地址：http://localhost:8010/taxi/batchTask/10，消息数据如下

显示OK说明消息已经发送到了kafka中

4.2.5 数据选择4.2.51 kafka数据查看

在load页面选择kafka，进行数据摄取模式选择

4.2.5.2 选择数据源

在这里输入ZK的地址以及需要选择数据的topic

116.62.213.90:10903,116.62.213.90:10904

4.2.5.3 加载数据

点击apply应用配置，设置加载数据源

4.2.6 数据源规范配置4.2.6.1 设置时间列

json 选择器被选中后，点击 Next：Parse time 进入下一步来决定您的主时间列。

因为我们的时间列有两个创建时间以及打车时间，我们配置时间列为trvelDate