oozie 上的 Avro Map-Reduce答案

【问题标题】：Avro Map-Reduce on oozieoozie 上的 Avro Map-Reduce
【发布时间】：2012-06-17 03:12:56
【问题描述】：

我一直在尝试在 oozie 上运行 Avro map-reduce。我在 workflow.xml 中指定了 mapper 和 reducer 类，并提供了其他配置。但它给出了一个

java.lang.RunTime Exception - class mr.sales.avro.etl.SalesMapper not org.apache.hadoop.mapred.Mapper

直接在 hadoop 集群（而不是通过 oozie）上运行相同的作业会完成并提供所需的输出。因此，我似乎可能缺少一些 oozie 配置。我从异常中猜想的是，oozie 要求映射器是 org.apache.hadoop.mapred.Mapper 的子类，但 Avro 映射器具有不同的签名 - 它们扩展了 org.apache.avro.mapred.AvroMapper，这可能是错误的原因。

所以我的问题是如何配置 oozie 工作流/属性文件以允许它运行 Avro map-reduce 作业。

【问题讨论】：

标签： serialization hadoop mapreduce avro oozie

【解决方案1】：

您也可以发布您的映射器和减速器类吗？我的 oozie 工作流程运行良好，但 o/p 文件不是 .avro 文件。这是我的工作流程：

<workflow-app name='sample-wf' xmlns="uri:oozie:workflow:0.2">
<start to='start_here'/>
<action name='start_here'>
    <map-reduce>
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <prepare>
            <delete path="${nameNode}/user/hadoop/${workFlowRoot}/final-output-data"/>
        </prepare>
        <configuration>

            <property>
                <name>mapred.job.queue.name</name>
                <value>${queueName}</value>
            </property>
            <property>
                  <name>mapred.reducer.new-api</name>
                  <value>true</value>
                </property>
                <property>
                  <name>mapred.mapper.new-api</name>
                  <value>true</value>
                </property>
            <property>
                <name>mapred.input.dir</name>
                <value>/user/hadoop/${workFlowRoot}/input-data</value>
            </property>
            <property>
                <name>mapred.output.dir</name>
                <value>/user/hadoop/${workFlowRoot}/final-output-data</value>
            </property>


            <property>
                <name>mapreduce.mapper.class</name>
                <value>org.apache.avro.mapred.HadoopMapper</value>
            </property>
            <property>
                <name>mapreduce.reducer.class</name>
                <value>org.apache.avro.mapred.HadoopReducer</value>
            </property>
            <property>
                <name>avro.mapper</name>
                <value>com.flipkart.flap.data.batch.mapred.TestAvro$CFDetectionMapper</value>
            </property>
            <property>
                <name>avro.reducer</name>
                <value>com.flipkart.flap.data.batch.mapred.TestAvro$CFDetectionReducer</value>
            </property>
            <property>
                <name>mapreduce.input.format.class</name>
                <value>org.apache.avro.mapreduce.AvroKeyInputFormat</value>
            </property>
            <property>
                <name>avro.schema.input.key</name>
                <value>{... schema ...}</value>
            </property>
           
            <property>
                <name>mapreduce.mapoutput.key.class</name>
                <value>org.apache.hadoop.io.AvroKey</value>
            </property>
            <property>
                <name>avro.map.output.schema.key</name>
                <value>{... schema ...}</value>
            </property>

            
            <property>
                <name>mapreduce.mapoutput.value.class</name>
                <value>org.apache.hadoop.io.Text</value>
            </property>
             <property>
                <name>mapreduce.output.format.class</name>
                <value>org.apache.avro.mapred.AvroKeyValueOutputFormat</value>
            </property>
            <property>
                <name>mapreduce.output.key.class</name>
                <value>org.apache.avro.mapred.AvroKey</value>
            </property>

            <property>
                <name>mapreduce.output.value.class</name>
                <value>org.apache.avro.mapred.AvroValue</value>
            </property>
           
            
            <property>
                <name>avro.schema.output.key</name>
                <value>{ ....   schema .... }</value>
            </property>
             <property>
                <name>avro.schema.output.value</name>
                <value>"string"</value>
            </property>
            <property>
                <name>mapreduce.output.key.comparator.class</name>
                <value>org.apache.avro.mapred.AvroKeyComparator</value>
            </property>
            <property>
                <name>io.serializations</name>
                <value>org.apache.hadoop.io.serializer.WritableSerialization,org.apache.avro.mapred.AvroSerialization
                </value>
            </property>
        </configuration>
    </map-reduce>
    <ok to='end'/>
    <error to='fail'/>
</action>
<kill name='fail'>
    <message>MapReduce failed, error message[$sf:errorMessage(sf:lastErrorNode())}]</message>
</kill>
<end name='end'/>
</workflow-app>

我的 mapper 和 reducer 是这样定义的：

public static class CFDetectionMapper extends
                Mapper<AvroKey<AdClickFraudSignalsEntity>, NullWritable, AvroKey<AdClickFraudSignalsEntity>, Text> {}

 public static class CFDetectionReducer extends
               Reducer<AvroKey<AdClickFraudSignalsEntity>, Text, AvroKey<AdClickFraudSignalsEntity>, AvroValue<CharSequence>>

【讨论】：

【解决方案2】：

这周我也遇到了同样的问题。这是我的 workflow.xml（已修改）：

<workflow-app name='sample-wf' xmlns="uri:oozie:workflow:0.2">
<start to='start_here'/>
<action name='start_here'>
    <map-reduce>
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <prepare>
            <delete path="${nameNode}/user/${wf:user()}/output"/>
        </prepare>
        <configuration>
            <property>
                <name>mapred.input.dir</name>
                <value>/user/${wf:user()}/input</value>
            </property>
            <property>
                <name>mapred.output.dir</name>
                <value>/user/${wf:user()}/output</value>
            </property>
            <property>
                <name>mapred.mapper.class</name>
                <value>org.apache.avro.mapred.HadoopMapper</value>
            </property>
            <property>
                <name>mapred.reducer.class</name>
                <value>org.apache.avro.mapred.HadoopReducer</value>
            </property>
            <property>
                <name>avro.mapper</name>
                <value>package.for.my.Mapper</value>
            </property>
            <property>
                <name>avro.reducer</name>
                <value>package.for.my.Reducer</value>
            </property>
            <property>
                <name>mapred.input.format.class</name>
                <value>org.apache.avro.mapred.AvroUtf8InputFormat</value>
            </property>
            <property>
                <name>mapred.output.format.class</name>
                <value>org.apache.avro.mapred.AvroOutputFormat</value>
            </property>
            <property>
                <name>mapred.output.key.class</name>
                <value>org.apache.avro.mapred.AvroWrapper</value>
            </property>
            <property>
                <name>mapred.mapoutput.key.class</name>
                <value>org.apache.avro.mapred.AvroKey</value>
            </property>
            <property>
                <name>mapred.mapoutput.value.class</name>
                <value>org.apache.avro.mapred.AvroValue</value>
            </property>
            <property>
                <name>avro.map.output.schema</name>
                <value>{put your schema here from job.xml via manual run}</value>
            </property>
            <property>
                <name>avro.input.schema</name>
                <value>"string"</value>
            </property>
            <property>
                <name>avro.output.schema</name>
                <value>{put your schema here from job.xml via manual run}</value>
            </property>
            <property>
                <name>mapred.output.key.comparator.class</name>
                <value>org.apache.avro.mapred.AvroKeyComparator</value>
            </property>
            <property>
                <name>io.serializations</name>
                <value>org.apache.hadoop.io.serializer.WritableSerialization,org.apache.avro.mapred.AvroSerialization</value>
            </property>
        </configuration>
    </map-reduce>
    <ok to='end'/>
    <error to='fail'/>
</action>
<kill name='fail'>
    <message>MapReduce failed, error message[$sf:errorMessage(sf:lastErrorNode())}]</message>
</kill>
<end name='end'/>

您可能需要根据 map-reduce 作业的输入和输出进行更多修改。

【讨论】：

【解决方案3】：

使用 AVRO，您需要配置一些额外的属性：

org.apache.avro.mapred.HadoopMapper 是您需要设置的实际映射器类（它实现了 Mapper 接口）
avro.mapper 属性应该命名你的 SalesMapper 类

combiner 和 reducer 还有其他属性 - 检查 AvroJob 源和实用方法。

另一种方法是检查您手动提交的作业中的 job.xml，并将相关配置属性复制到您的 oozie workflow.xml

【讨论】：

感谢您的回复。我可以通过 oozie 运行该作业，但输出似乎不是二进制 avro 格式 - 它看起来像一个文本文件，其中包含 org.apache.avro.mapred.AvroWrapper@448765ee 类型的条目错了，如果有？因为在我手动提交的作业中，输出看起来不同。我已经在 workflow.xml 中指定了 outputcehma
我错过了指定的 output.format 参数。当我指定它时它起作用了。