Oozie worflow with avro - output 是一个损坏的 avro 文件

Question

当我手动运行映射作业时，它会生成一个有效的 avro 文件，扩展名为 . avro 扩展名。但是当我在oozie工作流程中编写它时，它会生成一个文本文件，这是一个损坏的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>

我的映射器和化简器定义如下：

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>>
       
             

你能告诉我这里出了什么问题吗？

Answer 1

您使用了一些错误的属性名称：

<property>
  <name>mapreduce.mapoutput.key.class</name>
  <value>org.apache.hadoop.io.AvroKey</value>
</property>
[...]
<property>
  <name>mapreduce.mapoutput.value.class</name>
  <value>org.apache.hadoop.io.Text</value>
</property>

应该是：

<property> 
  <name>mapreduce.map.output.key.class</name>
  <value>org.apache.hadoop.io.AvroKey</value>
</property>
[...]
<property>
  <name>mapreduce.map.output.value.class</name>
  <value>org.apache.hadoop.io.Text</value>
</property>

（请注意添加的点）。对于早期的映射名称，这曾经是不同的，但现在是这样 - 请参阅 http://hadoop.apache.org/docs/r2.5.2/hadoop-project-dist/hadoop-common/DeprecatedProperties.html。