Druid 使用 Kafka 将数据载入到 Kafka

将数据载入到 Kafka

现在让我们为我们的主题运行一个生成器（producer），然后向主题中发送一些数据！

在你的 Druid 目录中，运行下面的命令：

代码语言：javascript复制

cd quickstart/tutorial
gunzip -c wikiticker-2015-09-12-sampled.json.gz > wikiticker-2015-09-12-sampled.json

在你的 Kafka 的安装目录中，运行下面的命令。请将 {PATH_TO_DRUID} 替换为 Druid 的安装目录：

代码语言：javascript复制

export KAFKA_OPTS="-Dfile.encoding=UTF-8"
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic wikipedia < {PATH_TO_DRUID}/quickstart/tutorial/wikiticker-2015-09-12-sampled.json

上面的控制台命令将会把示例消息载入到 Kafka 的 wikipedia 主题。现在我们将会使用 Druid 的 Kafka 索引服务（indexing service）来将我们加载到 Kafka 中的消息导入到 Druid 中。

使用数据加载器（data loader）来加载数据

在 URL 中导航到 localhost:8888 页面，然后在控制台的顶部单击Load data。

选择 Apache Kafka 然后单击 Connect data。

输入 Kafka 的服务器地址为 localhost:9092 然后选择 wikipedia 为主题。

然后单击 Apply。请确定你在界面中看到的数据只正确的。

一旦数据被载入后，你可以单击按钮 “Next: Parse data” 来进行下一步的操作。

Druid 的数据加载器将会为需要加载的数据确定正确的处理器。在本用例中，我们成功的确定了需要处理的数据格式为 json 格式。你可以在本页面中选择不同的数据处理器，通过选择不同的数据处理器，能够帮你更好的了解 Druid 是如何帮助你处理数据的。

当 json 格式的数据处理器被选择后，单击 Next: Parse time 来进行入下一个界面，在这个界面中你需要确定 timestamp 主键字段的的列。

Druid 要求所有数据必须有一个 timestamp 的主键字段（这个主键字段被定义和存储在 __time）中。如果你需要导入的数据没有时间字段的话，那么请选择 Constant value。在我们现在的示例中，数据载入器确定 time 字段是唯一可以被用来作为数据时间字段的数据。

单击 Next: ... 2 次，来跳过 Transform 和 Filter 步骤。针对本教程来说，你并不需要对导入时间进行换行，所以你不需要调整转换（Transform）和过滤器（Filter）的配置。

配置摘要（schema）是你对 dimensions 和 metrics 在导入数据的时候配置的地方。这个界面显示的是当我们对数据在 Druid 中进行导入的时候，数据是如何在 Druid 中进行存储和表现的。因为我们提交的数据集非常小，因此我们可以关闭回滚（rollup），Rollup 的开关将不会在这个时候显示来供你选择。

如果你对当前的配置满意的话，单击 Next 来进入 Partition 步骤。在这个步骤中你可以定义数据是如何在段中进行分区的。