java - Flume Twitter Agent 生成的 Avro 文本文件未在 Java 中读取

Question

无法使用 Flume 推特代理读取和解析流推特数据创建的文件，既不使用 Java 也不使用 Avro 工具。我的需求是将avro格式转换成JSON格式。

当使用任何一种方法时，我得到异常:org.apache.avro.AvroRuntimeException: java.io.IOException: Block size invalid or too large for this implementation: -40

我在伪节点集群中使用 Hadoop vanilla 配置，hadoop 版本是 2.7.1

Flume版本为1.6.0

twitter 代理的 flume 配置文件和解析 avro 文件的 java 代码附在下面:

TwitterAgent.sources=Twitter
TwitterAgent.channels=MemChannel
TwitterAgent.sinks=HDFS
TwitterAgent.sources.Twitter.type=org.apache.flume.source.twitter.TwitterSource
TwitterAgent.sources.Twitter.channels=MemChannel

TwitterAgent.sources.Twitter.consumerKey=xxxxxxxxxxxxxx
TwitterAgent.sources.Twitter.consumerSecret=xxxxxxxxxxxxxxxx
TwitterAgent.sources.Twitter.accessToken=xxxxxxxxxxxxxxxx
TwitterAgent.sources.Twitter.accessTokenSecret=xxxxxxxxxxxxxx

TwitterAgent.sources.Twitter.keywords=Modi,PMO,Narendra Modi,BJP

TwitterAgent.sinks.HDFS.channel=MemChannel
TwitterAgent.sinks.HDFS.type=hdfs
TwitterAgent.sinks.HDFS.hdfs.path=hdfs://localhost:9000/user/ashish/Twitter_Data
TwitterAgent.sinks.HDFS.hdfs.fileType=DataStream
TwitterAgent.sinks.HDFS.hdfs.writeformat=Text
TwitterAgent.sinks.HDFS.hdfs.batchSize=100
TwitterAgent.sinks.HDFS.hdfs.rollSize=0
TwitterAgent.sinks.HDFS.hdfs.rollCount=10
TwitterAgent.sinks.HDFS.hdfs.rollInterval=30
TwitterAgent.channels.MemChannel.type=memory
TwitterAgent.channels.MemChannel.capacity=10000
TwitterAgent.channels.MemChannel.transactionCapacity=100

---

import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.FileReader;
import org.apache.avro.file.SeekableInput;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.mapred.FsInput;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import java.io.IOException;

public class AvroReader {

    public static void main(String[] args) throws IOException {
        Path path = new Path("hdfs://localhost:9000/user/ashish/Twitter_Data/FlumeData.1449656815028");
        Configuration config = new Configuration();
        SeekableInput input = new FsInput(path, config);
        DatumReader<GenericRecord> reader = new GenericDatumReader<>();
        FileReader<GenericRecord> fileReader = DataFileReader.openReader(input, reader);

        for (GenericRecord datum : fileReader) {
            System.out.println("value = " + datum);
        }

        fileReader.close();
    }
}

我得到的异常堆栈跟踪是:

2015-12-09 17:48:19,291 WARN  [main] util.NativeCodeLoader (NativeCodeLoader.java:<clinit>(62)) - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    value = {"id": "674535686809120768", "user_friends_count": 1260, "user_location": "ユウサリ", "user_description": "「テガミバチ」に登場するザジのbotです。追加してほしい言葉などの希望があればＤＭでお願いします。リムーブする際はブロックでお願いします。", "user_statuses_count": 47762, "user_followers_count": 1153, "user_name": "ザジ", "user_screen_name": "zazie_bot", "created_at": "2015-12-09T15:56:54Z", "text": "@ill_akane_bot お前、なんか、\u2026すっげー楽しそうだな\u2026", "retweet_count": 0, "retweeted": false, "in_reply_to_user_id": 204695477, "source": "<a href=\"http:\/\/twittbot.net\/\" rel=\"nofollow\">twittbot.net<\/a>", "in_reply_to_status_id": 674535430423887872, "media_url_https": null, "expanded_url": null}
    Exception in thread "main" org.apache.avro.AvroRuntimeException: java.io.IOException: Block size invalid or too large for this implementation: -40
        at org.apache.avro.file.DataFileStream.hasNextBlock(DataFileStream.java:275)
        at org.apache.avro.file.DataFileStream.hasNext(DataFileStream.java:197)
        at avro.AvroReader.main(AvroReader.java:24)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:497)
        at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
    Caused by: java.io.IOException: Block size invalid or too large for this implementation: -40
        at org.apache.avro.file.DataFileStream.hasNextBlock(DataFileStream.java:266)
        ... 7 more

我是否还需要提供 Avro 架构才能正确读取 Avro 文件，如果是的话在哪里？

Answer 1

我也遇到了这个问题。虽然我可以看到你的数据文件已经不存在了。我检查了这个我的数据文件，应该和你的一样。

我发现我的数据文件已经是一个 avro 容器文件，这意味着它有它的模式和数据。

我得到的 avro 文件非常错误，因为它应该只包含一个包含 avro 模式的头，但实际上它的文件中有多个头。

另外，tweets 已经是 JSON 格式，为什么 flume 将它们转换成 avro 格式？