使用Flume采集Twitter流数据时遭遇twitter4J Jar文件错误求助

问题原因 #

• Twitter API v1.1流端点已废弃：Twitter已停用v1.1版本的statuses/sample.json流API端点，当前必须使用Twitter API v2的对应端点（如/2/tweets/sample/stream）。
• twitter4J版本过时：Flume 1.11.0默认依赖的twitter4j 4.0.7仅支持Twitter API v1.1，无法适配v2的接口规范与认证方式。
• 认证方式不兼容：Twitter API v2要求使用Bearer Token或OAuth 2.0认证，原Flume TwitterSource使用的OAuth 1.0a认证已无法访问废弃的v1.1端点。

解决方案 #

方案1：替换为支持Twitter API v2的自定义Flume Source #

由于官方Flume TwitterSource已停止维护，需使用适配v2的自定义Source，步骤如下：

1. 添加依赖库
   在Flume的lib目录下添加以下依赖：

• twitter-api-java-sdk（Twitter官方v2 SDK）
• jackson系列JSON解析依赖

2. 编写自定义TwitterSource（核心示例）

public class TwitterV2Source extends AbstractSource implements Configurable {
    private TwitterCredentials credentials;
    private TwitterStreamingClient streamingClient;

    @Override
    public void configure(Context context) {
        // 读取配置中的Bearer Token
        String bearerToken = context.getString("bearerToken");
        credentials = new TwitterCredentials.BearerTokenCredentials(bearerToken);
    }

    @Override
    public void start() {
        try {
            streamingClient = new TwitterStreamingClient(credentials);
            // 添加推文监听逻辑
            streamingClient.addTweetListener(tweet -> {
                // 将推文转为Flume Event并发送到Channel
                Event event = EventBuilder.withBody(tweet.toString().getBytes(StandardCharsets.UTF_8));
                getChannelProcessor().processEvent(event);
            });
            // 设置关键词过滤规则
            List<Rule> rules = Collections.singletonList(
                new Rule("AWS Marketplace OR GCP Marketplace OR Azure Marketplace")
            );
            streamingClient.addRules(rules);
            // 启动采样流
            streamingClient.sampleStream();
        } catch (TwitterException e) {
            getLogger().error("Twitter stream connection failed", e);
        }
    }

    @Override
    public void stop() {
        if (streamingClient != null) {
            streamingClient.close();
        }
    }
}

3. 修改Flume配置文件
   更新flume_project.conf中的Source配置：

TwitterAgent.sources.Twitter.type = com.your.package.TwitterV2Source
TwitterAgent.sources.Twitter.channels = MemChannel
TwitterAgent.sources.Twitter.bearerToken = YOUR_TWITTER_API_V2_BEARER_TOKEN
TwitterAgent.sources.Twitter.keywords = AWS Marketplace, GCP Marketplace, Azure Marketplace

方案2：改用Logstash采集Twitter流（替代Flume） #

如果不想自定义Flume Source，可直接使用支持API v2的Logstash Twitter插件：

1. 编写Logstash配置文件
   创建twitter-logstash.conf：

input {
  twitter {
    bearer_token => "YOUR_TWITTER_API_V2_BEARER_TOKEN"
    endpoints => { sample_stream => "tweets/sample/stream" }
    tweet_fields => ["id", "text", "created_at"]
  }
}

filter {
  mutate {
    add_field => { "topic" => "fp_mc" }
  }
}

output {
  kafka {
    bootstrap_servers => "fp_mc-kafka-1:9092"
    topic_id => "fp_mc"
  }
}

2. 启动Logstash
   在Docker容器中执行：

docker exec -it your-logstash-container logstash -f /docker_share/twitter-logstash.conf

方案3：临时兼容（仅适用于保留v1.1权限的账号） #

如果你的Twitter开发者账号仍保留v1.1访问权限，可尝试：

• 替换Flume lib目录下的twitter4j jar包为最新版本（4.1.2+）
• 确认OAuth 1.0a凭证正确，且账号拥有v1.1流API的访问权限

验证结果 #

启动Kafka消费者验证数据是否正常接收：

docker exec -i fp_mc-kafka-2 kafka-console-consumer.sh \
    --bootstrap-server fp_mc-kafka-1:9092 \
    --topic fp_mc \
    --from-beginning

内容的提问来源于stack exchange，提问作者netdevmike