Skip to content

add PrimitiveTypes support for CSVSerDe #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chutium wants to merge 3 commits into
base: master
Choose a base branch
from
Open

add PrimitiveTypes support for CSVSerDe #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
22c667c
add PrimitiveTypes support for CSVSerDe
chutium Sep 1, 2014
4186676
check empty string before parsing numbers
chutium Oct 6, 2014
2bea36b
fix EOL...
chutium Oct 6, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 106 additions & 43 deletions src/main/java/com/bizo/hive/serde/csv/CSVSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,71 +23,71 @@
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;

import au.com.bytecode.opencsv.CSVReader;
import au.com.bytecode.opencsv.CSVWriter;


/**
* CSVSerde uses opencsv (http://opencsv.sourceforge.net/) to serialize/deserialize columns as CSV.
*
*
* @author Larry Ogrodnek <ogrodnek@gmail.com>
*/
public final class CSVSerde implements SerDe {

private ObjectInspector inspector;
private String[] outputFields;
private int numCols;
private List<String> row;

private List<Object> row;
private List<TypeInfo> columnTypes;

private char separatorChar;
private char quoteChar;
private char escapeChar;



@Override
public void initialize(final Configuration conf, final Properties tbl) throws SerDeException {
final List<String> columnNames = Arrays.asList(tbl.getProperty(Constants.LIST_COLUMNS).split(","));
final List<TypeInfo> columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(tbl.getProperty(Constants.LIST_COLUMN_TYPES));
columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(tbl.getProperty(Constants.LIST_COLUMN_TYPES));

numCols = columnNames.size();

final List<ObjectInspector> columnOIs = new ArrayList<ObjectInspector>(numCols);

for (int i=0; i< numCols; i++) {
columnOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
columnOIs.add(TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(columnTypes.get(i)));
}
this.inspector = ObjectInspectorFactory.getStandardStructObjectInspector(columnNames, columnOIs);
this.outputFields = new String[numCols];
row = new ArrayList<String>(numCols);

inspector = ObjectInspectorFactory.getStandardStructObjectInspector(columnNames, columnOIs);
outputFields = new String[numCols];
row = new ArrayList<Object>(numCols);

for (int i=0; i< numCols; i++) {
row.add(null);
}

separatorChar = getProperty(tbl, "separatorChar", CSVWriter.DEFAULT_SEPARATOR);
quoteChar = getProperty(tbl, "quoteChar", CSVWriter.DEFAULT_QUOTE_CHARACTER);
escapeChar = getProperty(tbl, "escapeChar", CSVWriter.DEFAULT_ESCAPE_CHARACTER);
}

private final char getProperty(final Properties tbl, final String property, final char def) {
final String val = tbl.getProperty(property);

if (val != null) {
return val.charAt(0);
}

return def;
}

@Override
public Writable serialize(Object obj, ObjectInspector objInspector) throws SerDeException {
final StructObjectInspector outputRowOI = (StructObjectInspector) objInspector;
final List<? extends StructField> outputFieldRefs = outputRowOI.getAllStructFieldRefs();

if (outputFieldRefs.size() != numCols) {
throw new SerDeException("Cannot serialize the object because there are "
+ outputFieldRefs.size() + " fields but the table has " + numCols + " columns.");
Expand All @@ -97,45 +97,108 @@ public Writable serialize(Object obj, ObjectInspector objInspector) throws SerDe
for (int c = 0; c < numCols; c++) {
final Object field = outputRowOI.getStructFieldData(obj, outputFieldRefs.get(c));
final ObjectInspector fieldOI = outputFieldRefs.get(c).getFieldObjectInspector();

// The data must be of type String
// TODO: need to add switch case too
final StringObjectInspector fieldStringOI = (StringObjectInspector) fieldOI;

// Convert the field to Java class String, because objects of String type
// can be stored in String, Text, or some other classes.
outputFields[c] = fieldStringOI.getPrimitiveJavaObject(field);
}

final StringWriter writer = new StringWriter();
final CSVWriter csv = newWriter(writer, separatorChar, quoteChar, escapeChar);

try {
csv.writeNext(outputFields);
csv.close();

return new Text(writer.toString());
} catch (final IOException ioe) {
throw new SerDeException(ioe);
}
}
}

@Override
public Object deserialize(final Writable blob) throws SerDeException {
Text rowText = (Text) blob;

CSVReader csv = null;
try {
csv = newReader(new CharArrayReader(rowText.toString().toCharArray()), separatorChar, quoteChar, escapeChar);
csv = newReader(new CharArrayReader(rowText.toString().toCharArray()), separatorChar, quoteChar, escapeChar);
final String[] read = csv.readNext();

for (int i=0; i< numCols; i++) {
if (read != null && i < read.length) {
row.set(i, read[i]);
} else {
row.set(i, null);
if (columnTypes.get(i).getCategory() == ObjectInspector.Category.PRIMITIVE) {
PrimitiveTypeInfo columnType = (PrimitiveTypeInfo) columnTypes.get(i);
if (read != null && i < read.length) {
String val = read[i].trim();
switch (columnType.getPrimitiveCategory()) {
case BOOLEAN:
if (val.length() == 0)
row.set(i, null);
else
row.set(i, Boolean.valueOf(val));
break;
case BYTE:
if (val.length() == 0)
row.set(i, null);
else
row.set(i, Byte.valueOf(val));
break;
case DOUBLE:
if (val.length() == 0)
row.set(i, 0);
else
row.set(i, Double.valueOf(val));
break;
case FLOAT:
if (val.length() == 0)
row.set(i, 0);
else
row.set(i, Float.valueOf(val));
break;
case INT:
if (val.length() == 0)
row.set(i, 0);
else
row.set(i, Integer.valueOf(val));
break;
case LONG:
if (val.length() == 0)
row.set(i, 0);
else
row.set(i, Long.valueOf(val));
break;
case SHORT:
if (val.length() == 0)
row.set(i, 0);
else
row.set(i, Short.valueOf(val));
break;
case TIMESTAMP:
if (val.length() == 0)
row.set(i, null);
else
row.set(i, java.sql.Timestamp.valueOf(val));
break;
case STRING:
row.set(i, val);
break;
case VOID:
row.set(i, null);
break;
case UNKNOWN:
default:
throw new RuntimeException("Unknown PrimitiveType");
}
} else {
row.set(i, null);
}
}
}

return row;
} catch (final Exception e) {
throw new SerDeException(e);
Expand All @@ -149,22 +212,22 @@ public Object deserialize(final Writable blob) throws SerDeException {
}
}
}

private CSVReader newReader(final Reader reader, char separator, char quote, char escape) {
// CSVReader will throw an exception if any of separator, quote, or escape is the same, but
// CSVReader will throw an exception if any of separator, quote, or escape is the same, but
// the CSV format specifies that the escape character and quote char are the same... very weird
if (CSVWriter.DEFAULT_ESCAPE_CHARACTER == escape) {
return new CSVReader(reader, separator, quote);
} else {
return new CSVReader(reader, separator, quote, escape);
return new CSVReader(reader, separator, quote, escape);
}
}

private CSVWriter newWriter(final Writer writer, char separator, char quote, char escape) {
if (CSVWriter.DEFAULT_ESCAPE_CHARACTER == escape) {
return new CSVWriter(writer, separator, quote, "");
} else {
return new CSVWriter(writer, separator, quote, escape, "");
return new CSVWriter(writer, separator, quote, escape, "");
}
}

Expand All @@ -177,7 +240,7 @@ public ObjectInspector getObjectInspector() throws SerDeException {
public Class<? extends Writable> getSerializedClass() {
return Text.class;
}

public SerDeStats getSerDeStats() {
return null;
}
Expand Down
58 changes: 34 additions & 24 deletions src/test/java/com/bizo/hive/serde/csv/CSVSerdeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,54 +12,64 @@

public final class CSVSerdeTest {
private final CSVSerde csv = new CSVSerde();
final Properties props = new Properties();


final Properties new_props = new Properties();
final Properties props = new Properties();

@Before
public void setup() throws Exception {
new_props.put(Constants.LIST_COLUMNS, "a,b,c,d,e,f,g,h,i,j");
new_props.put(Constants.LIST_COLUMN_TYPES, "string,string,int,bigint,float,double,timestamp,timestamp,void,int");
props.put(Constants.LIST_COLUMNS, "a,b,c");
props.put(Constants.LIST_COLUMN_TYPES, "string,string,string");
props.put(Constants.LIST_COLUMN_TYPES, "string,string,int");
}

@Test
public void testDeserialize() throws Exception {
csv.initialize(null, props);
final Text in = new Text("hello,\"yes, okay\",1");
final List<String> row = (List<String>) csv.deserialize(in);
csv.initialize(null, new_props);
final Text in = new Text("hello,\"yes, okay\",1,\"1409576461919\",100.05,\"1024.19002010\",\"2014-04-14 00:03:49\",\"2014-04-14 00:09:42.228000\",,");

final List<Object> row = (List<Object>) csv.deserialize(in);

assertEquals("hello", row.get(0));
assertEquals("yes, okay", row.get(1));
assertEquals("1", row.get(2));
assertEquals(1, row.get(2));
assertEquals(1409576461919L, row.get(3));
assertEquals(Float.valueOf("100.05"), row.get(4));
assertEquals((Double) 1024.19002010, row.get(5));
assertEquals(java.sql.Timestamp.valueOf("2014-04-14 00:03:49"), row.get(6));
assertEquals(java.sql.Timestamp.valueOf("2014-04-14 00:09:42.228"), row.get(7));
assertEquals(null, row.get(8));
assertEquals(0, row.get(9));
}



@Test
public void testDeserializeCustomSeparators() throws Exception {
props.put("separatorChar", "\t");
props.put("quoteChar", "'");

csv.initialize(null, props);

final Text in = new Text("hello\t'yes\tokay'\t1");
final List<String> row = (List<String>) csv.deserialize(in);
final List<Object> row = (List<Object>) csv.deserialize(in);

assertEquals("hello", row.get(0));
assertEquals("yes\tokay", row.get(1));
assertEquals("1", row.get(2));
assertEquals("yes\tokay", row.get(1));
assertEquals(1, row.get(2));
}

@Test
public void testDeserializeCustomEscape() throws Exception {
props.put("quoteChar", "'");
props.put("escapeChar", "\\");

csv.initialize(null, props);

final Text in = new Text("hello,'yes\\'okay',1");
final List<String> row = (List<String>) csv.deserialize(in);
final List<Object> row = (List<Object>) csv.deserialize(in);

assertEquals("hello", row.get(0));
assertEquals("yes'okay", row.get(1));
assertEquals("1", row.get(2));
}
assertEquals(1, row.get(2));
}
}