如何使用 java 解析水槽事件（Twitter 源）

Posted 2023-02-16

【中文标题】如何使用 java 解析水槽事件（Twitter 源）

【英文标题】：How to parse flume events with java (Twitter Source)

【发布时间】：2019-05-14 14:20:11

【问题描述】：

我正在尝试使用 Flume 获取 twitter 事件并将它们存储在我创建的树数据结构中，以便聚合这些事件。

我可以在本地机器上使用来自 Java 的 Flume 节点应用程序启动 Flume 代理。在我的配置文件中，作为一个例子，我指定了一个文件roll sink，并且数据成功写入了一个文件中。

这是我用来启动代理的 Java 程序。

import org.apache.flume.node.Application;
import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;

public class StreamObserver

    static Logger logger = Logger.getLogger(StreamObserver.class);

    public static void main (String [] args)




        BasicConfigurator.configure();
        Application.main(args);


        //System.out.println(args[0]);
    

使用程序参数：

agent -n agentName -f flumeConfFile

这是我的配置文件：

# Name the components on this agent
t_agent.sources = TwitterSrc
t_agent.sinks = Lsink
t_agent.channels = Lchannel

# Describe/configure the source

t_agent.sources.TwitterSrc.type = org.apache.flume.source.twitter.TwitterSource

t_agent.sources.TwitterSrc.consumerKey = **********************
t_agent.sources.TwitterSrc.consumerSecret = *********************
t_agent.sources.TwitterSrc.accessToken = *****************
t_agent.sources.TwitterSrc.accessTokenSecret = *****************
t_agent.sources.TwitterSrc.maxBatchDurationMillis = 200


t_agent.sources.TwitterSrc.keywords = data



# Describe the sink
t_agent.sinks.Lsink.type = file_roll
t_agent.sinks.Lsink.channel = Lchannel
t_agent.sinks.Lsink.sink.directory = destDirectory/
t_agent.sinks.Lsink.fileHeader = true

# Use a channel which buffers events in memory
t_agent.channels.Lchannel.type = memory
t_agent.channels.Lchannel.capacity = 1000
t_agent.channels.Lchannel.transactionCapacity = 100


# Bind the source and sink to the channel
t_agent.sources.TwitterSrc.channels = Lchannel
t_agent.sinks.Lsink.channel = Lchannel

正如我所说，代理正常启动，事件被写入本地文件。

但我想知道是否有办法从我的 Java 程序中解析事件。

创建的文件是二进制格式，我也想知道是否有办法将它们转换为 ascii 格式，以便我检查结果。

注意，我使用的是 IntelliJ IDEA。

【参考方案1】：

你可能想调查一下 flink 是否适合你的需求

https://ci.apache.org/projects/flink/flink-docs-release-1.9/