如何使用 Flume 在 HDFS 中正确插入 JSON

Question

我正在使用Flume 中的HTTPSource 来接收json 格式的POST 事件，如下所示：

{"username":"xyz","password":"123"}

我的问题是：我是否必须修改事件的来源（我的意思是将 JSON 发送到 Flume 的那个）所以 JSON 具有以下格式：

[{
  "headers" : {
             "timestamp" : "434324343",
             "host" : "random_host.example.com"
             },
  "body" : "{"username":"xyz","password":"123"}"
}]

这是最好的方法吗？或者我可以在其他任何地方修改它？

flume agent 的 conf 文件是：

## Componentes
SomeAgent.sources = SomeHTTP
SomeAgent.channels = MemChannel
SomeAgent.sinks = SomeHDFS

## Fuente e Interceptores
SomeAgent.sources.SomeHTTP.type = http
SomeAgent.sources.SomeHTTP.port = 5140
SomeAgent.sources.SomeHTTP.handler = org.apache.flume.source.http.JSONHandler
SomeAgent.sources.SomeHTTP.channels = MemChannel
SomeAgent.sources.SomeHTTP.interceptors = i1 i2

## Interceptores
SomeAgent.sources.SomeHTTP.interceptors.i1.type = timestamp
SomeAgent.sources.SomeHTTP.interceptors.i2.type = host
SomeAgent.sources.SomeHTTP.interceptors.i2.hostHeader = hostname

## Canal
SomeAgent.channels.MemChannel.type = memory
SomeAgent.channels.MemChannel.capacity = 10000
SomeAgent.channels.MemChannel.transactionCapacity = 1000

## Sumidero
SomeAgent.sinks.SomeHDFS.type = hdfs
SomeAgent.sinks.SomeHDFS.channel = MemChannel
SomeAgent.sinks.SomeHDFS.hdfs.path = /raw/logs/%Y-%m-%d
SomeAgent.sinks.SomeHDFS.hdfs.fileType = DataStream
SomeAgent.sinks.SomeHDFS.hdfs.filePrefix = SomeLogs-
SomeAgent.sinks.SomeHDFS.hdfs.writeFormat = Text
SomeAgent.sinks.SomeHDFS.hdfs.batchSize = 100
SomeAgent.sinks.SomeHDFS.hdfs.rollSize = 0
SomeAgent.sinks.SomeHDFS.hdfs.rollCount = 10000
SomeAgent.sinks.SomeHDFS.hdfs.rollInterval = 600
SomeAgent.sinks.SomeHDFS.hdfs.useLocalTimeStamp = true

运行cat 的hadoop fs

$ hadoop fs -ls -R /raw/logs/somes
drwxr-xr-x   - flume-agent supergroup          0 2015-06-16 12:43 /raw/logs/arquimedes/2015-06-16
-rw-r--r--   3 flume-agent supergroup       3814 2015-06-16 12:33 /raw/logs/arquimedes/2015-06-16/SomeLogs.1434471803369
-rw-r--r--   3 flume-agent supergroup       3719 2015-06-16 12:43 /raw/logs/arquimedes/2015-06-16/SomeLogs.1434472404774


$ hadoop fs -cat /raw/logs/somes/2015-06-16/SomeLogs.1434471803369 | head




$

（你看对了，空行）

如果我现在查看文件（例如使用HUE 的二进制视图）：

0000000:    0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a   ................
0000010:    0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a   ................
0000020:    0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a 0a   ................

Answer 1

如果我理解得很好，您希望同时序列化数据和标头。在这种情况下，您不必修改数据源，而是使用一些标准 Flume 元素并为 HDFS 创建自定义序列化程序。

第一步是实现 Flume 创建所需的 JSON 结构，即 headers+body。 Flume 可以为你做这件事，只需在你的 HTTPSource 中使用JSONHandler，这样：

a1.sources = r1
a1.sources.r1.hnadler = org.apache.flume.source.http.JSONHandler

事实上，没有必要配置 JSON 处理程序，因为它是 HTTPSource 的默认处理程序。

然后，同时使用Timestamp Interceptor 和Host Interceptor 以添加所需的标题。唯一的技巧是 Flume 代理必须与发送者进程在同一台机器上运行，以便截获的主机与发送者的主机相同：

a1.sources.r1.interceptors = i1 i2
a1.sources.r1.interceptors.i1.type = timestamp
a1.sources.r1.interceptors.i2.type = host
a1.sources.r1.interceptors.i2.hostHeader = hostname

此时，您将拥有所需的事件。尽管如此，HDFS 的标准序列化程序只保存正文，而不保存标题。因此，创建一个实现org.apache.flume.serialization.EventSerializer 的自定义序列化程序。配置为：

a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.serializer = my_custom_serializer

HTH

Answer 2

@frb 发布的答案是正确的，唯一缺少的一点是 JSON 生成器必须发送 body 部分（我必须承认/抱怨 docs 在那一点），所以，发布json的正确方式是

[body:"{'username':'xyz','password':'123'}"]

请注意，json 的数据现在是一个字符串。

通过此更改，json 现在在 hdfs 中可见。

Answer 3

使用默认 JSONHandler 的 Flume HTTPSource 期望将 JSON 表示 [{ headers: ..., body: ... }] 的完整 Flume 事件列表提交到端点；要创建一个可以接受像{"username":"xyz", "password":"123"} 这样的裸应用程序级结构的代理端点，您可以使用实现HTTPSourceHandler 的替代类覆盖处理程序；请参阅 JSONHandler 来源 - 内容不多。

public List<Event> getEvents(HttpServletRequest request) throws ...

在自定义 JSONHandler 中，您还可以根据 HTTP 请求向事件添加标头，例如源 IP、User-Agent 等（拦截器不会为此提供上下文）。此时您可能想要验证应用程序提供的 JSON（尽管默认处理程序不会）。

尽管您已经发现，您可以只传递 [{body: ...}] 部分，但如果您想防止生成器为事件注入标头，这样的自定义处理程序也很有用。 p>