使用 Flume 反序列化 Json 文件并沉入 HDFS答案

【问题标题】：Deserializing Json file and sink into HDFS using flume使用 Flume 反序列化 Json 文件并沉入 HDFS
【发布时间】：2014-12-28 16:27:46
【问题描述】：

我有一个所有 json 文件都存在的假脱机目录，传入的文件将每秒添加到该目录中，我必须反序列化传入的 json 文件并获取所需字段并将其附加到 HDFS 目录中。

我所做的是创建了一个 Flume conf 文件，其中将 spooling 目录中的文件作为源，并使用 1 个 Sink 将 json 文件直接放入 HDFS。

我必须在 Sink 之前将这个 json 转换为结构格式并将其放入 HDFS。最重要的是，它不是推特数据。而且我必须实现纯粹的 Flume。

我使用下面的水槽配置来完成工作：

agent_slave_1.channels.fileChannel1_1.type = file 
agent_slave_1.channels.fileChannel1_1.capacity = 200000
agent_slave_1.channels.fileChannel1_1.transactionCapacity = 1000
agent_slave_1.sources.source1_1.type = spooldir

agent_slave_1.sources.source1_1.spoolDir = /home/cloudera/runs/
agent_slave_1.sources.source1_1.fileHeader = false
agent_slave_1.sources.source1_1.fileSuffix = .COMPLETED
agent_slave_1.sinks.hdfs-sink1_1.type = hdfs
agent_slave_1.sinks.hdfs-sink1_1.hdfs.path =hdfs://localhost.localdomain:8020/user/cloudera/runs_scored/
agent_slave_1.sinks.hdfs-sink1_1.hdfs.batchSize = 1000
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollSize = 268435456
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollInterval = 0
agent_slave_1.sinks.hdfs-sink1_1.hdfs.rollCount = 50000000
agent_slave_1.sinks.hdfs-sink1_1.hdfs.writeFormat=Text

agent_slave_1.sinks.hdfs-sink1_1.hdfsfileType = DataStream
agent_slave_1.sources.source1_1.channels = fileChannel1_1
agent_slave_1.sinks.hdfs-sink1_1.channel = fileChannel1_1

agent_slave_1.sinks =  hdfs-sink1_1
agent_slave_1.sources = source1_1
agent_slave_1.channels = fileChannel1_1

但我不知道如何使用反序列化器。

有人可以帮助我了解如何反序列化传入的 Json 文件吗？如果我需要用java编写任何代码，请帮助我，我需要使用什么接口？如果可能的话，给一些提示。

【问题讨论】：

你找到任何答案了吗？请分享。
没有。我也在寻找同样的东西。

标签： java json hadoop flume flume-ng

【解决方案1】：

最好的猜测是编写一个自定义拦截器，将您的 JSON 转换为所需的 HDFS 格式。它还具有填充头文件的好处，这些头文件可以在您的 hdfs 路径中使用。

以下是如何配置拦截器：

agent_slave_1.sources.source1_1.interceptors = my_intercptor
agent_slave_1.sources.source1_1.interceptors.my_intercptor.type = com.mycompany.MyInteceptor

这个类看起来像这样：

public class MyInteceptor implements Interceptor, Interceptor.Builder {

    private MyInteceptor interceptor;

    @Override
    public void initialize() {


    }

    @Override
    public Event intercept(Event event) {
        String bjson = event.getBody()));
        // decode your json, e.g. Jackson
        MyDecodedJsonObject record; // pseudo class
        event.getHeaders().put("timestamp", record.getTimestamp().toString());
        String newBody = record.getA() + "\t" + record.getB();
        event.setBody(newBody.getBytes())
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {

        for (Iterator<Event> iterator = events.iterator(); iterator.hasNext();) {
            Event next = intercept(iterator.next());
            if (next == null) {
                iterator.remove();
            }
        }
        return events;
    }

    @Override
    public void close() {


    }

    @Override
    public Interceptor build() {
        return interceptor;
    }

    @Override
    public void configure(Context context) {

        interceptor = new MyInteceptor();
    }

}

别忘了把这个类打包到一个jar里，放到flume的lib目录下。

【讨论】：