Python scripts for avro conversion using Hadoop Streaming … here is a solution to the problem.
Python scripts for avro conversion using Hadoop Streaming
I
have a 10 GB input file and I’m trying to convert it to avro using a python hadoop stream, it works successfully but I can’t read the output using the avro reader.
It gives the “utf8” codec unable to decode byte 0xb4 in position 13924: Invalid starting byte.
The problem here is that I use the standard output for the mapper output of the hadoop stream, and if I use the filename and use the script locally, the avro output is readable.
Any ideas, how to solve this problem? I think the problem is dealing with keys/values in streaming….
hadoop jar /opt/cloudera/parcels/CDH/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming.jar \
-input "xxx.txt" \
-mapper "/opt/anaconda/anaconda21/bin/python mapper.py x.avsc" \
-reducer NONE \
-output "xxxxx" -file "mapper.py" \
-lazyOutput \
-file "x.avsc"
The mapper script is
import sys
import re
import os
from avro import schema, datafile
import avro.io as io
import StringIO
schema_str = open("xxxxx.avsc", 'r').read()
SCHEMA = schema.parse(schema_str)
rec_writer = io. DatumWriter(SCHEMA)
df_writer = datafile. DataFileWriter(sys.stdout, rec_writer, SCHEMA,)
header = []
for field in SCHEMA.fields:
header.append(field.name)
for line in sys.stdin:
fields = line.rstrip().split("\x01")
data = dict(zip(header, fields))
try:
df_writer.append(data)
except Exception, e:
print "failed with data: %s" % str(data)
print str(e)
df_writer.close()
Solution
This problem can finally be solved. Use the output format class and leave the avro binary transformation to it. In the streaming mapper, you only need to emit a JSON record.
hadoop jar /opt/cloudera/parcels/CDH/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming.jar \
-libjars avro-json-1.2.jar \
-jobconf output.schema.url=hdfs:///x.avsc \
-input "xxxxx" \
-mapper "/opt/anaconda/anaconda21/bin/python mapper.py x.avsc" \
-reducer NONE \
-output "/xxxxx" \
-outputformat com.cloudera.science.avro.streaming.AvroAsJSONOutputFormat \
-lazyOutput \
-file "mapper.py" \
-file "x.avsc"
Here is mapper.py
import sys
from avro import schema
import json
schema_str = open("xxxxx.avsc", 'r').read()
SCHEMA = schema.parse(schema_str)
header = []
for field in SCHEMA.fields:
header.append(field.name)
for line in sys.stdin:
fields = line.rstrip().split("\x01")
data = dict(zip(header, fields))
try:
print >> sys.stdout, json.dumps(data, encoding='ISO-8859-1')
except Exception, e:
print "failed with data: %s" % str(data)
print str(e)