EMR Druid 探索(二)

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: EMR Druid 探索(二) EMR Druid 上文介绍了 Druid 的特点、使用场景以及性能。EMR 在 3.11.0 引入了 Druid,并专门推出了一种新的集群类型:Druid 集群。在具体使用时,Druid 集群可以与 Hadoop 集群结合,以 HDFS 集群作为 deep storage 的存储,以 YARN 作为批量索引的计算引擎。

EMR Druid 探索(二)

EMR Druid

上文介绍了 Druid 的特点、使用场景以及性能。EMR 在 3.11.0 引入了 Druid,并专门推出了一种新的集群类型:Druid 集群。在具体使用时,Druid 集群可以与 Hadoop 集群结合,以 HDFS 集群作为 deep storage 的存储,以 YARN 作为批量索引的计算引擎。Druid 集群亦可以与 Kafka 集群结合,将 Kafka 作为实时数据的来源。用户亦可用 Druid 摄取来自于 Spark Streaming 或者 Storm 的实时数据。

EMR 在 Druid 基础上也做了一些很有用的扩展,比如,

  • 支持 deep storage 到 OSS(一个很有用的功能,不必担心 HDFS 集群变动带来的数据危险),以及将 OSS 文件作为批量索引的数据来源
  • 支持 metadata 到 RDS,跟上一条结合在一起,Druid 集群升级就会变得非常方便
  • 在用户侧 EMR Druid 集成了 Superset 这一 BI 工具,可以方便地查询 Druid 数据,而不用写复杂的 DSL
  • 方便地扩容、缩容(针对 task 节点)
  • 丰富的指标监控以及告警规则、坏节点迁移,提供了高可靠的运维保障
  • 支持高安全、HA 等

示例

下面的流程展示了如何创建一个 Druid 集群,并进行批量索引。

首先,选择 Druid 集群类型。Druid 集群有几个必选组件,此外 HDFS、YARN 作为可选组件。在这里我们建议将这两个组件选上,一方面,Druid deep storage 到 OSS 的功能是借助于 HDFS 实现的,如果选择了 HDFS 用户不用做任何配置,如果没有选择的话,用户需要自己添加 HDFS 的配置文件,另一方面,选择这两个组件可以方便的在 Druid 集群内部进行批量索引测试,如果不需要,可以选择关闭他们。

1_create

下一步是选择节点类型和机器配置。Druid 本身不是一个纯内存计算系统,但是仍然会使用大量的内存。Druid 对内存的使用有两方面,一是计算过程中的数据,二是索引文件的 mapping。如果您的数据量很大,建议您选择大内存机型。8g 内存机型仅推荐作为测试机型。对于 core 节点,如果您的数据量很大,建议选择 d1 机型,这种机器本身配有多个本地磁盘,容量大,速度快,Druid 可以缓存更多 segments 到本地盘,而且 segments 加载速度会更快。

2_choosevm

选择好机器,点击下一步,填写集群名称,

3_name

再下一步、确认,等待一段时间,集群就创建好了。进入集群管理页面查看一下,

4_druid

下面我们将展示一个批量索引,并将索引的结果存储到 OSS 上。为此,首先在 OSS 上(用同一个账户)创建一个 oss://bucket/path,比如 “oss://emr-druid-cn-hangzhou/segments/”。之后,在 Druid 的配置页面,找到 druid.storage.storageDirectory,修改之。注意由于 OSS 访问是借助了 hdfs 的底层,因此 druid.storage.type 仍然为 “hdfs”,如图。

5_oss

依次点“保存”、“部署配置文件到主机”、“重启所有组件”。

组件重启成功之后,ssh 到 emr-header-1 节点,进入 druid 的安装目录 “/usr/lib/druid-current”,其下带有一份 quickstart 的演示示例。进入该目录,首先解压 “wikiticker-2015-09-12-sampled.json.gz”:

gunzip wikiticker-2015-09-12-sampled.json.gz

之后将该文件上传至 HDFS 的一个路径下面,这里我们选择 “/” 目录:

hdfs dfs -put wikiticker-2015-09-12-sampled.json /

也可以将该文件放置于某一 OSS 路径下面,EMR Druid 也是支持的。编辑 quickstart 下面的 wikiticker-index.json,其内容如下:

{
    "type" : "index_hadoop",
    "spec" : {
        "ioConfig" : {
            "type" : "hadoop",
            "inputSpec" : {
                "type" : "static",
                "paths" : "/wikiticker-2015-09-16-sampled.json"
            }
        },
        "dataSchema" : {
            "dataSource" : "wikiticker",
            "granularitySpec" : {
                "type" : "uniform",
                "segmentGranularity" : "day",
                "queryGranularity" : "none",
                "intervals" : ["2015-09-12/2015-09-13"]
            },
            "parser" : {
                "type" : "hadoopyString",
                "parseSpec" : {
                    "format" : "json",
                    "dimensionsSpec" : {
                        "dimensions" : [
                            "channel",
                            "cityName",
                            "comment",
                            "countryIsoCode",
                            "countryName",
                            "isAnonymous",
                            "isMinor",
                            "isNew",
                            "isRobot",
                            "isUnpatrolled",
                            "metroCode",
                            "namespace",
                            "page",
                            "regionIsoCode",
                            "regionName",
                            "user"
                        ]
                    },
                    "timestampSpec" : {
                        "format" : "auto",
                        "column" : "time"
                    }
                }
            },
            "metricsSpec" : [
                {
                    "name" : "count",
                    "type" : "count"
                },
                {
                    "name" : "added",
                    "type" : "longSum",
                    "fieldName" : "added"
                },
                {
                    "name" : "deleted",
                    "type" : "longSum",
                    "fieldName" : "deleted"
                },
                {
                    "name" : "delta",
                    "type" : "longSum",
                    "fieldName" : "delta"
                },
                {
                    "name" : "user_unique",
                    "type" : "hyperUnique",
                    "fieldName" : "user"
                }
            ]
        },
        "tuningConfig" : {
            "type" : "hadoop",
            "partitionsSpec" : {
                "type" : "hashed",
                "targetPartitionSize" : 5000000
            },
            "jobProperties" : {
                "mapreduce.job.classloader": "true"
            }
        }
    },
    "hadoopDependencyCoordinates": ["org.apache.hadoop:hadoop-client:2.7.2"]
}

注意索引文件路径为 HDFS(或 OSS)相应数据文件的路径,tuningConfig 里添加 "mapreduce.job.classloader": "true" 的 jobProperties,另外最后一行指定 Hadoop 版本为 2.7.2(EMR Hadoop 版本)。

运行如下命令提交索引作业

curl -XPOST -H 'Content-Type:application/json' -d @quickstart/wikiticker-index.json http://emr-header-1:18090/druid/indexer/v1/task

这时作业就被提交到了 Druid 集群,确切地说是 Overlord 节点。Overlord 节点会远程通知 Middlemanager 节点启动索引 task。该 task 依据索引类型做出处理,由于这里索引类型为 “index_hadoop”,该 task 会生成相应的 job 并提交到 YARN 执行。相应的执行情况可以通过页面查看:

首先打通到 emr-header-1 的 ssh 隧道

ssh -N -D 8157 root@<ip_of_emr-header-1>

然后启动设置了 proxy 的浏览器,proxy 指向 8157 端口,以 chrome 为例

open -n /Applications/Google\ Chrome.app --args --proxy-server="socks5://localhost:8157" --host-resolver-rules="MAP * 0.0.0.0 , EXCLUDE localhost" --user-data-dir=/tmp/

然后在地址栏输入“http://emr-header-1:18090/console.html,就能够看到相应的作业了,

6_overlord

同时打开 http://emr-header-1:8088/cluster/apps 能够看到相应的 Hadoop 作业:

7_hadoop

等待 Overlord 页面上显示索引成功,就能在 OSS 看到生成了相应的索引文件

8_ossdeep

运行一下查询命令(查询语句在 quickstart/wikiticker-top-pages.json)

curl -XPOST -H 'Content-Type:application/json' -d @quickstart/wikiticker-top-pages.json 'http://emr-header-1:18082/druid/v2/?pretty'

返回结果

[ {
  "timestamp" : "2015-09-12T00:46:58.771Z",
  "result" : [ {
    "edits" : 33,
    "page" : "Wikipedia:Vandalismusmeldung"
  }, {
    "edits" : 28,
    "page" : "User:Cyde/List of candidates for speedy deletion/Subpage"
  }, {
    "edits" : 27,
    "page" : "Jeremy Corbyn"
  }, {
    "edits" : 21,
    "page" : "Wikipedia:Administrators' noticeboard/Incidents"
  }, {
    "edits" : 20,
    "page" : "Flavia Pennetta"
  }, {
    "edits" : 18,
    "page" : "Total Drama Presents: The Ridonculous Race"
  },...
]

说明

  1. 上述示例中我们用到了 Druid 安装包自带的示例。事实上由于 Druid 接口均为 rest 接口,你可以在任何一台能够连接到 Druid 集群的机器上运行命令;
  2. Druid 的 rest api 使用起来有诸多不便,幸运的是 Druid 有对 SQL 的支持,虽然还不支持全部标准 SQL,但也足以应付多数场景了。EMR Druid 集成了 Superset 工具,可以方便的使用该工具对 Druid 数据进行查询和管理,具体使用方式见这里
  3. 实际应用中,Druid 更多的使用流的方式索引数据,如果要求数据不丢失,且只被消费一次,通常是流+批的形式。流的方式先摄入数据,保存一份批的数据晚上再处理一遍,确保数据不丢。如果数据要求不是很高,批可以不用。具体流的使用方式参见 EMR 文档Druid 官方文档或者 Tranquility 官方文档;
  4. 掌握必要的排错技巧是必要的,druid 索引排错指南见这里
相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
目录
相关文章
|
存储 SQL druid
EMR Druid 探索(一)
EMR Druid 探索(一) 什么是 Druid、Druid 使用场景 Druid 是 Metamarkets 公司(一家为在线媒体或广告公司提供数据分析服务的公司)推出的一个分布式内存实时分析系统,用于解决如何在大规模数据集下进行快速的、交互式的查询和分析。
3894 0
|
3月前
|
分布式计算 大数据 MaxCompute
EMR Remote Shuffle Service实践问题之阿里云RSS的开源计划内容如何解决
EMR Remote Shuffle Service实践问题之阿里云RSS的开源计划内容如何解决
|
3月前
|
分布式计算 测试技术 调度
EMR Remote Shuffle Service实践问题之集群中落地阿里云RSS如何解决
EMR Remote Shuffle Service实践问题之集群中落地阿里云RSS如何解决
|
1月前
|
SQL 存储 缓存
阿里云EMR StarRocks X Paimon创建 Streaming Lakehouse
本文介绍了阿里云EMR StarRocks在数据湖分析领域的应用,涵盖StarRocks的数据湖能力、如何构建基于Paimon的实时湖仓、StarRocks与Paimon的最新进展及未来规划。文章强调了StarRocks在极速统一、简单易用方面的优势,以及在数据湖分析加速、湖仓分层建模、冷热融合及全链路ETL等场景的应用。
253 2
阿里云EMR StarRocks X Paimon创建 Streaming Lakehouse
|
24天前
|
SQL 存储 缓存
降本60% ,阿里云 EMR StarRocks 全新发布存算分离版本
阿里云 EMR Serverless StarRocks 现已推出全新存算分离版本,该版本不仅基于开源 StarRocks 进行了全面优化,实现了存储与计算解耦架构,还在性能、弹性伸缩以及多计算组隔离能力方面取得了显著进展。
253 6
|
28天前
|
SQL 存储 缓存
阿里云EMR StarRocks X Paimon创建 Streaming Lakehouse
讲师焦明烨介绍了StarRocks的数据湖能力,如何使用阿里云EMR StarRocks构建基于Paimon的极速实时湖仓,StarRocks与Paimon的最新进展及未来规划。
119 3
|
2月前
|
SQL 分布式计算 Serverless
阿里云 EMR Serverless Spark 版正式开启商业化
阿里云 EMR Serverless Spark 版正式开启商业化,内置 Fusion Engine,100% 兼容开源 Spark 编程接口,相比于开源 Spark 性能提升300%;提供 Notebook 及 SQL 开发、调试、发布、调度、监控诊断等一站式数据开发体验!
146 3
阿里云 EMR Serverless Spark 版正式开启商业化
|
2月前
|
SQL 存储 NoSQL
阿里云 EMR StarRocks 在七猫的应用和实践
本文整理自七猫资深大数据架构师蒋乾老师在 《阿里云 x StarRocks:极速湖仓第二季—上海站》的分享。
253 2
|
3月前
|
存储 分布式计算 大数据
大数据革新在即,阿里云EMR如何布局DeltaLake引领行业潮流?
【8月更文挑战第26天】大数据时代,实时处理与分析能力对企业至关重要。Delta Lake 作为高性能、可靠且支持 ACID 事务的开源存储层,已成为业界焦点。阿里云 EMR 深度布局 Delta Lake,计划深化集成、强化数据安全、优化实时性能,并加强生态建设与社区贡献。通过与 Spark 的无缝对接及持续的技术创新,阿里云 EMR 致力于提供更高效、安全的数据湖解决方案,引领大数据处理领域的发展新方向。
49 3
|
3月前
|
存储 分布式计算 监控
揭秘阿里云EMR:如何巧妙降低你的数据湖成本,让大数据不再昂贵?
【8月更文挑战第26天】阿里云EMR是一种高效的大数据处理服务,助力企业优化数据湖的成本效益。它提供弹性计算资源,支持根据需求调整规模;兼容并优化了Hadoop、Spark等开源工具,提升性能同时降低资源消耗。借助DataWorks及Data Lake Formation等工具,EMR简化了数据湖构建与管理流程,实现了数据的统一化治理。此外,EMR还支持OSS、Table Store等多种存储选项,并配备监控优化工具,确保数据处理流程高效稳定。通过这些措施,EMR帮助企业显著降低了数据处理和存储成本。
127 3