数据库OLAP大数据后端【免费下载链接】druidApache Druid: a high performance real-time analytics database.项目地址https://gitcode.com/gh_mirrors/druid6/druid点击查看免费下载本篇技术指南以 Apache Druid 仓库中的社区扩展druid-influxdb-emitter为核心官方文档系统讲解如何将该扩展加载进 Druid、配置全部参数、理解其基于队列的异步发送模型并深入剖析 Druid 指标事件到 InfluxDB Line Protocol 的转换规则。读完本文你将能够在自己的 Druid 集群中启用该扩展把服务指标Service Metric Events稳定地投递到 InfluxDB 时序数据库供 Grafana 等可视化平台进一步分析与告警。扩展简介与工作流程druid-influxdb-emitter是一个社区贡献的 Druid Emitter 扩展作用是把 Druid 产生的监控指标通过 HTTP 协议发送到 InfluxDB 时序数据库。与 Druid 内置的 logging、http 等 emitter 相比它的目标消费者是 InfluxDB因此数据以 InfluxDB 原生的Line Protocol行协议文本格式投递。从源码实现看InfluxdbEmitter.java该扩展的工作流程分为四个阶段接收事件入队Druid 的监控框架每产生一个事件emit(Event event)方法被调用。该方法只处理ServiceMetricEvent类型的事件其余事件类型被忽略并将其放入一个有界阻塞队列LinkedBlockingQueue容量由maxQueueSize决定见 InfluxdbEmitter.java。定时批量转换start()启动一个单线程的ScheduledExecutorService按照flushDelay首次延迟和flushPeriod固定周期调度transformAndSendToInfluxdb()见 InfluxdbEmitter.java。全量冲刷发送transformAndSendToInfluxdb()一次性取出当前队列中的全部事件eventsQueue.size()决定取出数量逐个poll()逐条转换为 Line Protocol 后拼接再 POST 到 InfluxDB 的 HTTP 写接口见 InfluxdbEmitter.java。整个队列在一次发送中全部清空而不是逐条发送。关闭时兜底冲刷close()被调用进程关闭/优雅退出时会先调用flush()将队列中剩余事件全部发送随后关闭调度线程见 InfluxdbEmitter.java。注意该扩展目前只发送 Service Metric Events服务指标事件即 Druid metrics 文档中列举的那一类指标其他 feed如 alerts不会进入该队列。加载扩展druid-influxdb-emitter属于社区扩展Community Extension不随默认 Druid 发行包tarball打包。要使用它需要先把它下载安装到 Druid 的extensions目录再在common.runtime.properties的druid.extensions.loadList中加入扩展名。根据 扩展加载文档社区扩展通常可以通过pull-deps工具按 Maven 坐标下载。对于该扩展其 Maven groupId 为org.apache.druid.extensions.contribartifactId 为druid-influxdb-emitter版本为当前 Druid 稳定版本java \ -cp lib/* \ -Ddruid.extensions.directoryextensions \ -Ddruid.extensions.hadoopDependenciesDirhadoop-dependencies \ org.apache.druid.cli.Main tools pull-deps \ --no-default-hadoop \ -c org.apache.druid.extensions.contrib:druid-influxdb-emitter:DRUID_VERSION安装完成后在common.runtime.properties中加载扩展并启用该 emitterdruid.extensions.loadList[druid-influxdb-emitter, ...其他扩展] druid.emitterinfluxdb其中druid.emitterinfluxdb用于激活该 emitter 模块对应源码中的 InfluxdbEmitterModule.java其内部常量EMITTER_TYPE influxdbdruid.emitter的取值决定使用哪个 emitter 模块可参考 配置总览。配置参数详解该扩展的全部配置项都位于druid.emitter.influxdb前缀之下写入common.runtime.properties。下表完整列出了所有参数来自 官方文档属性说明是否必填默认值druid.emitter.influxdb.hostnameInfluxDB 服务器的主机名是无druid.emitter.influxdb.portInfluxDB 服务器端口否8086druid.emitter.influxdb.protocol发送指标使用的协议http或https二选一否httpdruid.emitter.influxdb.trustStorePathhttps 场景下使用的 trustStore 路径否无druid.emitter.influxdb.trustStoreTypehttps 场景下使用的 trustStore 类型否jks实际取java.security.KeyStore.getDefaultType()druid.emitter.influxdb.trustStorePasswordhttps 场景下使用的 trustStore 密码否无druid.emitter.influxdb.databaseNameInfluxDB 中的数据库名是无druid.emitter.influxdb.maxQueueSize存放待发送事件的队列容量否Integer.MAX_VALUE2^31-1druid.emitter.influxdb.flushPeriod每隔多少毫秒把事件队列解析为 Line Protocol 并 POST 到 InfluxDB否60000druid.emitter.influxdb.flushDelay定时任务首次执行前等待多少毫秒否60000druid.emitter.influxdb.influxdbUserName访问 InfluxDB 数据库的用户名是无druid.emitter.influxdb.influxdbPassword该用户的密码是无druid.emitter.influxdb.dimensionWhitelist允许作为 tag 写入 Line Protocol 的维度白名单否[dataSource,type,numMetrics,numDimensions,threshold,dimension,taskType,taskStatus,tier]默认值与源码验证上述默认值并非仅见于文档在配置类 InfluxdbEmitterConfig.java 中有直接对应的常量定义private static final int DEFAULT_PORT 8086; private static final int DEFAULT_QUEUE_SIZE Integer.MAX_VALUE; private static final int DEFAULT_FLUSH_PERIOD 60000; // milliseconds private static final ListString DEFAULT_DIMENSION_WHITELIST Arrays.asList( dataSource, type, numMetrics, numDimensions, threshold, dimension, taskType, taskStatus, tier); private static final String DEFAULT_PROTOCOL http;几个值得注意的细节hostname、databaseName、influxdbUserName、influxdbPassword四个字段在构造函数中通过Preconditions.checkNotNull强制校验缺失任何一个都会在启动阶段直接抛出异常配置类构造失败相关行为被 InfluxdbEmitterConfigTest.java 的testConfigWithNullHostname、testConfigWithNullInfluxdbUserName、testConfigWithNullInfluxdbPassword等用例覆盖。flushDelay未配置时默认值同样取 60000与flushPeriod相同即首次冲刷延迟 1 分钟。trustStoreType未配置时取KeyStore.getDefaultType()在绝大多数 JDK 上即jks与文档表格中标注的jks一致。port未配置时回退为 8086InfluxDB 默认 HTTP 端口对应测试testConfigWithNullPort。一个完整的配置示例# 加载扩展并启用 influxdb emitter druid.extensions.loadList[druid-influxdb-emitter] druid.emitterinfluxdb # InfluxDB 连接与认证 druid.emitter.influxdb.hostnameinfluxdb.example.com druid.emitter.influxdb.port8086 druid.emitter.influxdb.protocolhttp druid.emitter.influxdb.databaseNamedruid_metrics druid.emitter.influxdb.influxdbUserNamedruid_writer druid.emitter.influxdb.influxdbPasswordyour_password_here # 队列与冲刷策略可选以下为默认值 druid.emitter.influxdb.maxQueueSize2147483647 druid.emitter.influxdb.flushPeriod60000 druid.emitter.influxdb.flushDelay60000 # 维度白名单可选以下为默认值 druid.emitter.influxdb.dimensionWhitelist[dataSource,type,numMetrics,numDimensions,threshold,dimension,taskType,taskStatus,tier]参数调优建议基于实现机制flushPeriod / maxQueueSize 的取舍因为发送逻辑是整队列一次性清空maxQueueSize相当于高水位缓冲。若指标量大而flushPeriod过长队列可能积压较多事件单次 HTTP 请求体随之变大反之flushPeriod过短会增大对 InfluxDB 的请求频率。生产环境建议结合 Druid 侧指标量观察 InfluxDB 的写入负载后调整。HTTPS 场景protocolhttps时trustStorePath与trustStorePassword二者必须同时提供否则构造客户端时会抛出IllegalStateExceptionCant load TrustStore. Truststore path or password is not set.这一点在 InfluxdbEmitter.java 的buildInfluxdbClient()中实现并被InfluxdbEmitterTest的三个异常用例验证缺路径、缺密码、路径无效均抛异常。InfluxDB 侧前置要求启用认证与授权使用该扩展前必须在InfluxDB 服务端启用认证与授权authentication and authorization。因为 emitter 在构造写入请求时会把用户名密码直接拼进 HTTP 查询串POST /write?dbdatabaseNameuinfluxdbUserNamepinfluxdbPassword对应源码见 InfluxdbEmitter.java 的postToInflux()。若 InfluxDB 未开启认证该用户名/密码参数会被忽略但按照官方文档说明正确姿势仍是先在 InfluxDB 侧创建具备对应数据库写入权限的用户再配置到 Druid 中。Line Protocol 转换规则核心机制InfluxDB 通过 HTTP 写入数据时使用Line Protocol文本格式。其语法为measurement[,tag_keytag_value[,tag_keytag_value]] field_keyfield_value[,field_keyfield_value] [timestamp]其中timestamp为自 epoch 起的纳秒数。转换规则Druid 的 Service Metric Event 中metric字段是一个以/分隔的多段字符串例如query/cache/total/hits。扩展按照以下规则把它映射为 Line Protocol见 InfluxdbEmitter.java 的transformForInfluxSystems()measurement测量名druid_ metric 的第一段。例如query/cache/total/hits→druid_query。tags标签service事件中的 service 字段原样保留如druid/historicalmetricmetric 的中间段去掉首尾段用_拼接并冠以druid_前缀。例如query/cache/total/hits→druid_cache_total如果 metric 只有两段如query/time则没有中间段也就不生成 metric taghostname事件中的 host 字段去掉端口部分按:切分取第一段如historical001:8083→historical001白名单维度事件携带的用户维度user dims中凡是命中dimensionWhitelist的维度都会追加为 tag。field字段druid_ metric 的最后一段值为事件中的 value。例如query/cache/total/hits→druid_hits34787256。timestamp时间戳事件创建时间event.getCreatedTime()换算为纳秒毫秒值 × 1000000。官方文档示例一个由 Druid logging emitter 记录的典型服务指标事件Event [{feed:metrics,timestamp:2017-10-31T09:09:06.857Z,service:druid/historical,host:historical001:8083,version:0.11.0-SNAPSHOT,metric:query/cache/total/hits,value:34787256}]按上述规则转换后得到可直接 POST 到 InfluxDB 的字符串druid_query,servicedruid/historical,hostnamehistorical001,metricdruid_cache_total druid_hits34787256 1509440946857000000逐段拆解组成部分内容来源measurementdruid_querymetric 首段querytagservicedruid/historical事件 servicetaghostnamehistorical001事件 host 去掉端口tagmetricdruid_cache_totalmetric 中间段cache/total以_拼接并加druid_前缀fielddruid_hits34787256metric 末段hits 事件 valuetimestamp1509440946857000000事件时间戳的纳秒表示注意每条转换后的记录末尾会追加一个换行符\n源码中payload.append(StringUtils.format( %d\n, ...))多条事件拼接后形成多行 Line Protocol 主体一次性 POST。源码级验证测试用例即规范仓库中的单元测试 InfluxdbEmitterTest.java 用实际断言固化了上述规则是最直观的可执行文档testTransformForInfluxWithLongMetricmetric 为metric/te/st/value4 段期望输出druid_metric,servicedruid/historical,metricdruid_te_st,hostnamelocalhost,dataSourcetest_datasource druid_value1234 1509357600000000000同时验证了命中白名单的dataSource维度被追加为 tag而未在白名单中的nonWhiteListedDim维度被丢弃。testTransformForInfluxWithShortMetricmetric 为metric/time2 段期望输出druid_metric,servicedruid/historical,hostnamelocalhost druid_time1234 ...验证两段 metric 不产生 metric tag的规则。testMetricIsInDimensionWhitelist/testMetricIsInDefaultDimensionWhitelist分别验证自定义白名单与默认白名单下维度 tag 的生成行为。维度值清洗规则Line Protocol 的 tag 值中不允许出现点号.和空白字符因此 emitter 对写入的维度值做了统一清洗sanitize()方法使用正则[\s]|[.]连续空白或连续点号将所有匹配字符替换为下划线_见 InfluxdbEmitter.java 与 InfluxdbEmitter.java。维度白名单dimensionWhitelistDruid 的事件可能携带大量用户自定义维度user dims如果全部写入 Line Protocol会导致 InfluxDB 的 tag 基数爆炸显著影响写入性能与查询效率。因此该扩展引入了白名单机制事件中携带的维度只有当维度名在dimensionWhitelist中时才会作为 tag 追加到 Line Protocol白名单默认包含dataSource、type、numMetrics、numDimensions、threshold、dimension、taskType、taskStatus、tier可通过配置druid.emitter.influxdb.dimensionWhitelist覆盖默认值JSON 数组格式未配置时使用默认集合见 InfluxdbEmitterConfig.java。实际判断逻辑位于transformForInfluxSystems()for (String dimName : dimNames) { if (this.dimensionWhiteList.contains(dimName)) { tag.append(StringUtils.format(,%1$s%2$s, dimName, sanitize(String.valueOf(event.getUserDims().get(dimName))))); } }即遍历事件全部用户维度仅对白名单命中的维度执行追加 tag 值清洗两步操作。HTTPS 支持与 TrustStore 配置当druid.emitter.influxdb.protocolhttps时emitter 会使用 Apache HttpClient 构造一个带自定义SSLContext的客户端校验trustStorePath与trustStorePassword均已配置否则抛IllegalStateException从trustStorePath读取 KeyStore类型为trustStoreType默认jks用密码加载基于该 TrustStore 初始化TrustManagerFactory并构造 TLS 的SSLContext使用NoopHostnameVerifier跳过主机名校验构建 HttpClient见 InfluxdbEmitter.java。HTTPS 场景的最小配置示例druid.emitter.influxdb.protocolhttps druid.emitter.influxdb.trustStorePath/path/to/truststore.jks druid.emitter.influxdb.trustStoreTypejks druid.emitter.influxdb.trustStorePasswordtruststore_password若使用自签证书或内部 CA需要把对应的 CA 证书导入到上述 trustStore 中Druid 侧才能与 InfluxDB 完成 TLS 握手。常见问题排查要点启动即报错若hostname、databaseName、influxdbUserName、influxdbPassword缺失配置类构造失败进程启动失败日志会提示对应字段不可为 null。HTTPS 配置缺失protocolhttps但未同时提供 trustStore 路径与密码抛出IllegalStateException日志提示 Cant load TrustStore. Truststore path or password is not set.。写入失败postToInflux()中 POST 请求异常网络不通、认证失败、数据库不存在等会被捕获并记录 info 日志Failed to post events to InfluxDB.不会导致 Druid 进程崩溃但数据会丢失队列已被清空。因此务必提前确认 InfluxDB 侧数据库已创建、认证用户具备写入权限、网络可达。数据未出现确认druid.emitterinfluxdb已设置、扩展已加入druid.extensions.loadList并注意flushDelay/flushPeriod默认均为 60000ms首次冲刷发生在启动约 1 分钟后。小结druid-influxdb-emitter为 Druid 提供了一条通向 InfluxDB 的轻量指标通路事件先入有界队列再按固定周期整体转换为 InfluxDB Line Protocol 批量 POST关闭时兜底冲刷。本文从扩展加载、参数配置、InfluxDB 前置要求、Line Protocol 转换规则到 HTTPS 支持完整还原了该扩展的官方文档与源码实现。核心实现可继续阅读 InfluxdbEmitter.java、InfluxdbEmitterConfig.java 与 InfluxdbEmitterTest.java 三个文件其中测试用例对转换规则的逐字符断言是最值得信赖的行为规范。赞分享数据库OLAP大数据后端【免费下载链接】druidApache Druid: a high performance real-time analytics database.项目地址https://gitcode.com/gh_mirrors/druid6/druid点击查看免费下载相关推荐Apache Druid StatsD Emitter 扩展实战指南将 Druid 指标实时推送至 StatsD / StatsiteApache Druid StatsD Emitter 扩展实战指南将 Druid 指标实时推送至 StatsD / Statsite 本文以当前仓库中 st数据库数据分析OLAP大数据实时分析数据仓库后端Apache Druid Ambari Metrics Emitter 扩展将 Druid 指标接入 Ambari Metrics 监控体系Apache Druid Ambari Metrics Emitter 扩展将 Druid 指标接入 Ambari Metrics 监控体系 导读 本文围绕数据库数据分析OLAP大数据实时分析数据仓库后端Apache Druid Graphite Emitter 指南将 Druid 指标通过 Pickle 协议送入 Graphite CarbonApache Druid Graphite Emitter 指南将 Druid 指标通过 Pickle 协议送入 Graphite Carbon 本文以 Ap数据库数据分析OLAP大数据实时分析数据仓库后端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考