如何通过 Java 代码更改 Apache Flume 的配置文件？

Posted 2023-02-16

【中文标题】如何通过 Java 代码更改 Apache Flume 的配置文件？

【英文标题】：How to change configuration file of Apache flume through Java code?

【发布时间】：2018-04-03 20:37:23

【问题描述】：

我目前正在从事一个大数据项目，用于对 twitter 的热门话题进行情绪分析。跟着cloudera的教程，了解了如何通过flume将tweets发送到Hadoop。

http://blog.cloudera.com/blog/2012/09/analyzing-twitter-data-with-hadoop/

flume.conf：

# 获得 Apache 软件基金会 (ASF) 的一项许可 # 或更多贡献者许可协议。请参阅通知文件 # 随本作品分发以获取更多信息 # 关于版权所有权。 ASF 许可此文件 # 根据 Apache 许可证，版本 2.0（ ＃ “执照”）;除非合规，否则您不得使用此文件 # 使用许可证。您可以在以下网址获取许可证的副本 # # http://www.apache.org/licenses/LICENSE-2.0 # # 除非适用法律要求或书面同意， # 根据许可分发的软件分布在 #“原样”基础，不提供任何保证或条件 # 种类，明示或暗示。请参阅许可证 # 特定语言管理权限和限制 # 根据许可证。 # 配置文件需要定义源， # 通道和接收器。 # 每个代理都定义了源、通道和接收器， # 在这种情况下称为“TwitterAgent” TwitterAgent.sources = 推特 TwitterAgent.channels = MemChannel TwitterAgent.sinks = HDFS TwitterAgent.sources.Twitter.type = com.cloudera.flume.source.TwitterSource TwitterAgent.sources.Twitter.channels = MemChannel TwitterAgent.sources.Twitter.consumerKey = TwitterAgent.sources.Twitter.consumerSecret = TwitterAgent.sources.Twitter.accessToken = TwitterAgent.sources.Twitter.accessTokenSecret = TwitterAgent.sources.Twitter.keywords = hadoop、大数据、分析、大数据、cloudera、数据科学、数据科学家、商业智能、mapreduce、数据仓库、数据仓库、mahout、hbase、nosql、newsql、商业智能、云计算 TwitterAgent.sinks.HDFS.channel = MemChannel TwitterAgent.sinks.HDFS.type = hdfs TwitterAgent.sinks.HDFS.hdfs.path = hdfs://hadoop1:8020/user/flume/tweets/%Y/%m/%d/%H/ TwitterAgent.sinks.HDFS.hdfs.fileType = 数据流 TwitterAgent.sinks.HDFS.hdfs.writeFormat = 文本 TwitterAgent.sinks.HDFS.hdfs.batchSize = 1000 TwitterAgent.sinks.HDFS.hdfs.rollSize = 0 TwitterAgent.sinks.HDFS.hdfs.rollCount = 10000 TwitterAgent.channels.MemChannel.type = 内存 TwitterAgent.channels.MemChannel.capacity = 10000 TwitterAgent.channels.MemChannel.transactionCapacity = 100

现在要将其扩展到我的应用程序，我需要在 Flume 的配置文件中使用关键字部分来获取热门话题，我想出了 Java 代码来获取热门话题，但是我现在有一个问题，我不知道如何连接此代码到水槽配置文件或如何在关键字部分添加实时趋势主题的新文件。我在网上搜索了很多，因为我是这个领域的初学者，如果你提供一些信息或至少有一些其他的选择，这将有很大的帮助。

【问题讨论】：

如果不重启 Flume 代理就无法在运行时编辑配置

哦，是这样，那我会试着找到它。谢谢@cricket_007

【参考方案1】：

一个非常有趣的问题..！

我同意@cricket_007 的评论 - 无法在不重新启动 Flume 代理的情况下编辑配置。

我不能说太多，因为我还没有看到你的 Java 代码来获取热门话题的关键字。但是，根据您提供的信息，我可以想到一种替代方法（或者我应该说一种解决方法） - 但我自己还没有尝试过。

您可以像这样修改TwitterSource.java 类：

public void configure(Context context) 
consumerKey = context.getString(TwitterSourceConstants.CONSUMER_KEY_KEY);
consumerSecret = context.getString(TwitterSourceConstants.CONSUMER_SECRET_KEY);
accessToken = context.getString(TwitterSourceConstants.ACCESS_TOKEN_KEY);
accessTokenSecret = context.getString(TwitterSourceConstants.ACCESS_TOKEN_SECRET_KEY);

//MODIFY THE FOLLOWING PORTION
String keywordString = context.getString(TwitterSourceConstants.KEYWORDS_KEY, "");
if (keywordString.trim().length() == 0) 
    keywords = new String[0];
 else 
  keywords = keywordString.split(",");
  for (int i = 0; i < keywords.length; i++) 
    keywords[i] = keywords[i].trim();
  

//UNTIL THIS POINT

ConfigurationBuilder cb = new ConfigurationBuilder();
cb.setOAuthConsumerKey(consumerKey);
cb.setOAuthConsumerSecret(consumerSecret);
cb.setOAuthAccessToken(accessToken);
cb.setOAuthAccessTokenSecret(accessTokenSecret);
cb.setJSONStoreEnabled(true);
cb.setIncludeEntitiesEnabled(true);

twitterStream = new TwitterStreamFactory(cb.build()).getInstance(); 

我已经在上面的注释中添加了您正在初始化关键字字符串变量的位置 - 您可以调用您的 java 代码（我假设它是一种方法，您可以从中返回一个逗号分隔的关键字字符串）而不是提取this 来自 flume.conf 中可用的上下文（只需删除 context.getString() 部分）。

除此之外，只需从 flume.conf 中删除以下语句：

TwitterAgent.sources.Twitter.keywords = hadoop, big data, analytics, bigdata, cloudera, data science, data scientiest, business intelligence, mapreduce, data warehouse, data warehousing, mahout, hbase, nosql, newsql, businessintelligence, cloudcomputing

我希望这会有所帮助。

【讨论】：

我尝试按照此代码获取趋势 milindjagre.wordpress.com/2016/10/19/… 但未能构建包，因为 API getTrends() 可能会抛出 TwitterException，必须处理，但无法处理在可配置（）函数中，因为它是一个覆盖函数。为了获得趋势，我必须以某种方式处理异常，但由于覆盖我不能。

嗯，好的。我没有想到这一点。但是，即使您使用 try-catch 块或仅在 throws 声明中，您的构建也会失败吗？我只是用我自己的代码尝试了一个实验，并添加了一个虚拟的 try-catch 块，然后我就能够构建包了。在这里补充一点，我使用的是 AbstractEventDrivenSource 中的 doConfigure() 方法，而不是 Configurable 接口中的 configure() 方法。

谢谢，它通过添加一个 try catch 块来工作，并且我能够在 Hadoop 文件系统中获取有关趋势主题的推文的 Jason。现在我想对这条推文进行情感分析，你有什么建议吗？

我很高兴它对你有用。 :) 然而，在情绪分析方面，我自己还没有真正尝试过，但我观察到有很多关于这个主题的参考资料。由于您是在 Java 中执行此操作，因此这里有一个 API - lexalytics.com/support，它为此目的提供了一个 Java SDK。因此，如果您遇到任何问题，我建议您开始并发布另一个问题。这里的社区非常有帮助，您一定会成功的。