Avro Map-Reduce on oozie

Question

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

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作业。

Answer 1

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

• org.apache.avro.mapred.HadoopMapper是需要设置的实际映射器类(这实现了映射器interface)
• avro.mapper属性应将SalesMapper类命名为

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

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

Answer 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作业的输入和输出，您可能需要对其进行更多的修改。

Answer 3

你也可以发布你的mapper和reducer类吗？我的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>>

​