2

我的用例是我想将 Avro 数据从 Kafka 推送到 HDFS。Camus 似乎是正确的工具，但我无法使其工作。我是 camus 的新手，试图让 camus-example 工作， https://github.com/linkedin/camus

现在我正在尝试使 camus-example 工作。但是我仍然面临问题。

DummyLogKafkaProducerClient 的代码片段

package com.linkedin.camus.example.schemaregistry;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import com.linkedin.camus.etl.kafka.coders.KafkaAvroMessageEncoder;
import com.linkedin.camus.example.records.DummyLog;

public class DummyLogKafkaProducerClient {


    public static void main(String[] args) {

        Properties props = new Properties();

        props.put("metadata.broker.list", "localhost:6667");
        // props.put("serializer.class", "kafka.serializer.StringEncoder");
        // props.put("partitioner.class", "example.producer.SimplePartitioner");
        //props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, byte[]> producer = new Producer<String, byte[]>(config);

        KafkaAvroMessageEncoder encoder = get_DUMMY_LOG_Encoder();

        for (int i = 0; i < 500; i++) {
            KeyedMessage<String, byte[]> data = new KeyedMessage<String, byte[]>("DUMMY_LOG", encoder.toBytes(getDummyLog()));
            producer.send(data);

        }
    }

    public static DummyLog getDummyLog() {
        Random random = new Random();
        DummyLog dummyLog = DummyLog.newBuilder().build();
        dummyLog.setId(random.nextLong());
        dummyLog.setLogTime(new Date().getTime());
        Map<CharSequence, CharSequence> machoStuff = new HashMap<CharSequence, CharSequence>();
        machoStuff.put("macho1", "abcd");
        machoStuff.put("macho2", "xyz");
        dummyLog.setMuchoStuff(machoStuff);
        return dummyLog;
    }

    public static KafkaAvroMessageEncoder get_DUMMY_LOG_Encoder() {
        KafkaAvroMessageEncoder encoder = new KafkaAvroMessageEncoder("DUMMY_LOG", null);
        Properties props = new Properties();
        props.put(KafkaAvroMessageEncoder.KAFKA_MESSAGE_CODER_SCHEMA_REGISTRY_CLASS, "com.linkedin.camus.example.schemaregistry.DummySchemaRegistry");
        encoder.init(props, "DUMMY_LOG");
        return encoder;

    }
}

我还添加了默认无参数构造函数 ot DummySchemaRegistry，因为它给出了实例化异常

package com.linkedin.camus.example.schemaregistry;

import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;

import com.linkedin.camus.example.records.DummyLog;
import com.linkedin.camus.example.records.DummyLog2;
import com.linkedin.camus.schemaregistry.MemorySchemaRegistry;

/**
 * This is a little dummy registry that just uses a memory-backed schema registry to store two dummy Avro schemas. You
 * can use this with camus.properties
 */
public class DummySchemaRegistry extends MemorySchemaRegistry<Schema> {
    public DummySchemaRegistry(Configuration conf) {
        super();
        super.register("DUMMY_LOG", DummyLog.newBuilder().build().getSchema());
        super.register("DUMMY_LOG_2", DummyLog2.newBuilder().build()
                .getSchema());
    }
    public DummySchemaRegistry() {
        super();
        super.register("DUMMY_LOG", DummyLog.newBuilder().build().getSchema());
        super.register("DUMMY_LOG_2", DummyLog2.newBuilder().build().getSchema());
    }
}

在运行程序后我得到的异常跟踪下方

线程“主”com.linkedin.camus.coders.MessageEncoderException 中的异常：org.apache.avro.AvroRuntimeException：org.apache.avro.AvroRuntimeException：字段 id 类型：LONG pos:0 未设置并且在 com 中没有默认值。 linkedin.camus.etl.kafka.coders.KafkaAvroMessageEncoder.init(KafkaAvroMessageEncoder.java:55) 在 com.linkedin.camus.example.schemaregistry.DummyLogKafkaProducerClient.get_DUMMY_LOG_Encoder(DummyLogKafkaProducerClient.java:57) 在 com.linkedin.camus.example。 schemaregistry.DummyLogKafkaProducerClient.main(DummyLogKafkaProducerClient.java:32) 原因：org.apache.avro.AvroRuntimeException: org.apache.avro.AvroRuntimeException: Field id type:LONG pos:0 not set and has no default value at com.linkedin .camus.example.records.DummyLog$Builder.build(DummyLog.java:214) 在 com.linkedin.camus。sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) 在 sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl) 的 example.schemaregistry.DummySchemaRegistry.(DummySchemaRegistry.java:16) .java:45) 在 java.lang.reflect.Constructor.newInstance(Constructor.java:408) 在 java.lang.Class.newInstance(Class.java:438) 在 com.linkedin.camus.etl.kafka.coders。 KafkaAvroMessageEncoder.init(KafkaAvroMessageEncoder.java:52) ... 2 更多原因：org.apache.avro.AvroRuntimeException：字段 id 类型：LONG pos:0 未设置并且在 org.apache.avro.data 中没有默认值。 RecordBuilderBase.defaultValue(RecordBuilderBase.java:151) 在 com.linkedin.camus.example.records.DummyLog$Builder。构建（DummyLog.java:209）... 9 更多

4

4 回答 4

1

我想 camus 期望 Avro 模式具有默认值。我已将 dummyLog.avsc 更改为跟随并重新编译-

{ “namespace”：“com.linkedin.camus.example.records”，“type”：“record”，“name”：“DummyLog”，“doc”：“不那么重要的日志。”，“fields” ：[{“名称”：“id”，“类型”：“int”，“默认”：0}，{“名称”：“logTime”，“类型”：“int”，“默认”：0}] }

请让我知道这对你有没有用。

谢谢，安巴里什

于 2015-02-09T23:56:14.833 回答
0

您可以默认任何字符串或长字段，如下所示

  {"type":"record","name":"CounterData","namespace":"org.avro.usage.tutorial","fields":[{"name":"word","type":["string","null"]},{"name":"count","type":["long","null"]}]}
于 2015-03-24T05:14:56.977 回答
0

我遇到了这个问题，因为我正在像这样初始化注册表：

super.register("DUMMY_LOG_2", LogEvent.newBuilder().build().getSchema());

当我将其更改为：

super.register("logEventAvro", LogEvent.SCHEMA$);

这让我通过了例外。

我也用过加里的 com.linkedin.camus.etl.kafka.coders.AvroMessageDecoder。

我还发现这个博客（Alvin Jin 的笔记本）非常有用。它指出了您在使用 camus 示例时可能遇到的所有问题并解决了它！

于 2015-12-23T04:51:06.163 回答
0

Camus 不假设模式将具有默认值。我最近使用 camus 发现了同样的问题。实际上它在模式注册表中使用的方式在默认示例中是不正确的。我已经对 Camus 代码进行了一些修改，您可以查看https://github.com/chandanbansal/camus进行一些小改动以使其正常工作。他们没有 Avro 唱片的解码器。我也写过。

于 2015-09-30T10:53:34.247 回答