问题标签 [camus]

我对使用 LinkedIn 的 camus 很陌生，并且已经成功地将数据文件从 Kafka 写入 Hdfs。通常，我使用JsonStringMessagdecoder读取 JSON 并将其写入.dat文件使用StringRecordWriterProvider.

但是可以写入多种文件类型吗？

假设 jsonkafka如下：

现在我想John,Hi how are you?写入一个文件，同时user,message写入.meta同一位置的另一个文件（）。可能吗 ？

有什么方法可以防止camus增加偏移值，并使其始终读取先前的值？

以及如何为创建的文件提供自定义名称而不是一般格式，即topic.leaderId.partitionId.partitioner.encodePartition(context, key)？

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

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

DummyLogKafkaProducerClient 的代码片段

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

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

线程“主”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 更多

我下载了包含 camus jar 的 confluent 包，并按照在线说明在此处输入链接描述。

Hadoop 设置正确（这意味着我可以使用 hadoop fs -ls 命令和其他 hadoop jar 命令）。但是，当我尝试运行时

我收到“主”类未发现错误

“confluent-camus-1.0.jar”的路径是正确的（就在文件夹下面）。我没有启动kafka服务，只是为了尝试运行它。

有人遇到过类似的问题吗？

谢谢。

尝试使用包含运行映射器作业Oozie的工作流来运行协调器。协调器似乎每 20 分钟运行一次并启动工作流，但工作流将无限期地运行，即使独立运行的作业可以在几分钟内轻松完成。我认为该错误与我如何运行作业或如何传递参数有关，但我不确定如何调试它。这是代码：java actionCamus

/coord/job.properties

坐标/协调器.xml

/workflow/workflow.xml

SHADED jar 和 config.properties 位于 /workflow/lib/

我正在运行 HDP 2.2

协调员日志：

工作流日志：

我设置了一个 Oozie java 操作工作流，我计划使用 Oozie 协调器来安排它。java 操作运行一个 Camus 作业，我已将其 jar 和属性配置文件放在工作流/lib 目录中。关于如何将 -P 参数传递给此的任何想法？目前，我正在做这样的事情：

camusJobProperties 的样子

但工作流程似乎没有运行（卡在 PREP 上）。任何想法如何解决这一问题？

谢谢！

编辑：更正我的 nameNode URL 后，我可以看到我收到以下错误：

所以基本上我的问题是，当属性文件位于 HDFS 中（特别是在工作流/lib 目录中）时，我将如何传递属性文件参数

我刚开始接触加缪。我计划每小时运行一次加缪工作。我们每小时收到约 80000000 条消息（平均大小约 4KB）。

如何设置以下属性：

我无法清楚地弄清楚这些配置。我应该将天数设为 1 并将小时数属性设为 2 吗？camus 是如何提取数据的？我经常看到以下错误：

如何正确设置配置以每小时运行一次并避免该错误？

我刚从加缪开始。

我计划每隔一小时运行一次加缪。我们80000000每小时处理一次〜消息，平均消息大小为4KB（我们在 Kafka 中有一个主题）。

我第一次尝试使用10映射器，复制一个小时的数据需要大约 2 小时，它创建了 10 个大小约为 7GB 的文件。

然后我尝试300了映射器，它把时间缩短到了大约 1 小时。但它创建了 11 个文件。后来，我尝试使用150映射器，大约花了 30 分钟。

那么，我该如何选择其中的映射器数量呢？此外，我想在 hadoop 中创建更多文件，因为一种大小增长到 7GB。我必须检查什么配置？

我想使用 Kafka 将 apache 服务器日志加载到 hdfs。
创建主题：

跟踪apache访问日志目录：

在另一个终端（kafka bin）启动消费者：

camus.properties 文件配置为：

执行以下命令时出现错误：

以下是错误：

请建议可以做些什么来解决这个问题。提前致谢

深沉

我对 Camus 和 Hadoop 非常陌生，并且遇到了异常错误。我正在尝试将一些 avro 文件写入 hdfs，并不断收到以下错误块：

我查看了第 108 行com.linkedin.camus.etl.kafka.common.KafkaReader.getNext，发现它是这样的：MessageAndOffset msgAndOffset = messageIter.next();。

我正在使用io.confluent.camus.etl.kafka.coders.AvroMessageDecoder我的解码器和com.linkedin.camus.example.DummySchemaRegistry我的编码器。

在日志的末尾，我得到另一行，指示来自 hdfs 文件之一的错误：Error from file [hdfs://localhost:9000/user/username/exec/2015-12-17-17-05-25/errors-m-00000]. error-m-00000 文件包含一个有点可读的开头，但随后更改为无法辨认的字符串：

SEQ*com.linkedin.camus.etl.kafka.common.EtlKey5com.linkedin.camus.etl.kafka.common.ExceptionWritable*org.apache.hadoop.io.compress.DefaultCodec|Ò ∫±ß˝}pºHí$ò¸ ·:0 schemasQ∞ΔøÿxúïîÀN√0E7l‡+∫»¢lFMõ> á*êxU®™ËzÍmàc[ÆÕ„XÚÕÿqZ%@[ÿD±gÓô…¯∆üGœ¯Ç¿Q,·Úçë2ô'«hZL¿3ëSöXÿ5ê·ê„Sé‡ÇÖpÎS¬î4,...LËÕ¥Î{û}wFßáâ*M)>%&uZÑCfi“˚#rKÌÔ¡flÌu^Í%† B∂"Xa*•⁄0ÔQÕpùGzùidy&ñªkT...Å›Ô^≥-#0>›...ΔRG∫.ˇÅ¨«JÚ®sÃ≥Ö</em>¡\£Rîfi˚ßé<em>T≥D#%T8ãW® ÚµÌ∫4N˙©W∫©mst√—Ô嶥óhÓ$C~#S+Ñâ{ã ÇFL¡ßí⁄L´ÏíÙºÙΩ5wfÃjM¬∏_Äò5RØ£ Ë"Eeúÿëx{ÆÏ«{XW÷XM€O¨- C#É¡Òl•ù9§‰õö2ó:wɲ%Œ-N∫ˇbFXˆ∑:àá5fyQÑ'ö™:roõ1⁄5•≠≈˚yM0±ú?»ÃW◊.h≈I´êöNæ [û3

最后，根据时间报告，hadoop 作业似乎已经运行，但从未发生提交：

任何帮助或在哪里寻找解决这个问题的想法将不胜感激。谢谢你。