位置: 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

4.2.6.2 设置转换器

在这里可以新增虚拟列,将一个列的数据转换成另一个虚拟列,这里我们增加一个状态的虚拟列,来显示状态的中文名称我们定义 0:测试数据, 1:发起打车,2:排队中,3:司机接单,4:乘客上车,5:完成打车

我们使用case_simple来实现判断功能,更多判断功能参考

case_simple(status,0,'测试数据',1,'发起打车',2,'排队中',3,'司机接单',4,'完成打车','状态错误')

在这里我们新建了一个status_text的虚拟列来展示需要中文显示的列

配置年龄默认值,如果为空我们设置为25

nvl(age,25)

配置性别设置,我们需要设置为男女,0:男,1:女,如果为null,我们设置为男

case_simple(nvl(sex,0),0,'男',1,'女','男')

4.2.6.3 设置过滤器

这里可以设置过滤器,对于某些数据不展示,这里我们使用区间过滤器选择显示status>=1的数据,具体表达式可用参考

{ "type" : "bound", "dimension" : "status", "ordering": "numeric", "lower": "1", }

因为我们把数据是0的测试数据不显示了,所以只显示了一条数据为1的数据

4.2.6.4 配置schema

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

4.2.6.5 配置Partition

一旦对schema满意后,点击 Next 后进入 Partition 步骤,该步骤中可以调整数据如何划分为段文件的方式,因为我们打车一般按照小时来算的,我们设置为分区为``hour

4.2.6.6 配置拉取方式

这里设置kafka的拉取方式,主要设置偏移量的一些配置

​在 Tune 步骤中,将 Use earliest offset 设置为 True 非常重要,因为我们需要从流的开始位置消费数据。 其他没有任何需要更改的地方,进入到 Publish 步

4.5.7 提交任务4.2.7.1 发布数据

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

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

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

4.2.7.2 提交任务

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

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

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

4.2.7.3 查看数据源

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

4.2.7.4 查询数据

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

4.2.7.5 动态添加数据

发送一条数据到kafka

druid 查询数据,发现新的数据已经进来了

4.2.8 清理数据4.2.8.1 关闭集群# 进入impl安装目录cd /usr/local/imply/imply-2021.05-1# 关闭集群./bin/service --down

4.2.8.2 等待关闭服务

通过进程查看,查看服务是否已经关闭

ps -ef|grep druid

4.2.8.3 清理数据

通过删除druid软件包下的var目录的内容来重置集群状态

llrm -rf var

4.2.8.4 重新启动集群 nohup bin/supervise -c conf/supervise/quickstart.conf > logs/quickstart.log 2>&1 &4.2.8.5 查看数据源

登录后查看数据源,我们发现已经被重置了

本文由传智教育博学谷 - 狂野架构师教研团队发布如果本文对您有帮助,欢迎关注和点赞;如果您有任何建议也可留言评论或私信,您的支持是我坚持创作的动力转载请注明出处!

本文链接地址:https://www.jiuchutong.com/zhishi/304694.html 转载请保留说明!

上一篇:关于 background-image 渐变gradient()那些事!(关于我和鬼变成家人的那件事)

下一篇:[Python]解密pyc文件(python解密加密文件)

  • 新开网吧需要缴纳哪些税金
  • 个人经营所得税起征点是多少
  • 补提去年所得税会计分录怎么写
  • 餐饮流水账表格excel
  • 银行汇票用什么账户核算
  • 开票系统怎么切换到数字账户
  • 负数发票可以作废不
  • 出口货物不能退税的原因
  • 货车的折旧年限怎么算
  • 物业公司银行存款专款专怎么做分录
  • 已付款收不到发票怎么办
  • 公司电视机 摊销多少年
  • 退休人员工资可以提福利费吗
  • 房租收入如何缴税?
  • 是否退还投标文件
  • 其他业务收入记账凭证怎么写
  • 非正常损失进项税额怎么处理
  • 抵扣税款是什么意思呢
  • 专票可以当费用票吗
  • 税前列支的好处
  • 2018年个体工商户增值税起征点
  • 关于金融行业的经典句子
  • 危险废物处理原则
  • 增值税专票销售额比对(非差额企业)
  • 电子发票转收入怎么做为记账凭证?
  • 清算时未分配利润为负数的处理方法是什么呢?
  • 是否所有监控设备都录像
  • 即征即退的增值税计入其他收益吗
  • 锁定任务栏不管用
  • 甲方工程扣款如何处理
  • 进项税一定是专票吗
  • 蜻蜓翅膀特写怎么画
  • php中array_multisort对多维数组排序的方法
  • thinkphp5上传图片
  • 债权投资 科目
  • Python Module — OpenAI ChatGPT API
  • 移动端页设计
  • 缴纳残保金会计分录最新
  • 收到预付购货款
  • 利息进项税是否含增值税
  • 逾期的押金税率是多少
  • 应收账款逾期无法偿还
  • 结转材料成本差异所需科目
  • 固定资产改扩建和大修理的区别
  • 番茄开发票属于蔬菜吗?
  • 个税申报表中的年金是什么意思?
  • 公司申请破产后债务谁来还
  • 福利费超过多少交税
  • SQL Server解析XML数据的方法详解
  • mysql 局域网访问权限
  • 个人工资薪金如何零申报
  • 超市的库存商品是怎么记账的
  • 红冲作废怎么处理
  • 发票拍照打印出来可以报销吗法律
  • 个体工商户税务注销流程
  • 交易性金融资产属于流动资产
  • 因管理不善材料被盗应记啥科目
  • 原材料用于在建工程进项税可以抵扣吗
  • 现金流量少
  • 技术支持费用属于什么费用
  • 利息与资金占用费
  • 多种不同的 MySQL 的 SSL 配置
  • mysql %d
  • pe下修改windows下的盘符
  • ubuntu的终端在哪
  • Linux一键安装ftp
  • linux常用命令useradd
  • win7桌面壁纸自动更换关闭
  • linux中密码信息存在哪个文件
  • js时间范围
  • 通过node-mysql搭建Windows+Node.js+MySQL环境的教程
  • Ubuntu 16.04 64位中搭建Node.js开发环境教程
  • androidstudio 教程
  • javascript基础教程答案
  • JavaScript快速排序
  • python怎么定义
  • 企业代理申报还用自己申报吗
  • 八戒财税主要做什么
  • 税务机关支部活动方案
  • 江苏国税发票出库时间
  • 免责声明:网站部分图片文字素材来源于网络,如有侵权,请及时告知,我们会第一时间删除,谢谢! 邮箱:opceo@qq.com

    鄂ICP备2023003026号

    网站地图: 企业信息 工商信息 财税知识 网络常识 编程技术

    友情链接: 武汉网站建设