Merge pull request #2 from civitaspo/v0.0.2

v0.0.2

Author: civitaspo

CHENGELOG.md

@@ -1,3 +1,10 @@
+0.0.2 (2016-01-23)
+==================
+- incompatible change: option `column_name` is removed
+- add: option `column`
+- add: option `skip_if_null`
+- enhance: support json type
+
 0.0.1 (2016-01-22)
 ==================
 - first release.

README.md

@@ -8,14 +8,20 @@ Convert a record to jsonl.
 
 ## Configuration
 
-- **column_name**: json column name (string, default: `"json_payload"`)
+- **column**: output json column (optional)
+  - **name** (string, default: `"json_payload"`)
+  - **type** string or json (string, default: `"string"`)
+- **skip_if_null**: input column name list (array of string, default: `[]`)
 
 ## Example
 
 ```yaml
 filters:
   - type: to_json
-    column_name: json_column
+    column:
+      name: test
+      type: string
+    skip_if_null: [id]
 ```
 
 ## Run Example
@@ -25,9 +31,6 @@ $ ./gradlew classpath
 $ embulk run -I lib example/config.yml
 ```
 
-## TODO
-- support json type
-
 ## Build
 
 ```

build.gradle

@@ -13,7 +13,7 @@ configurations {
     provided
 }
 
-version = "0.0.1"
+version = "0.0.2"
 
 sourceCompatibility = 1.7
 targetCompatibility = 1.7

example/config.yml

@@ -15,5 +15,9 @@ in:
       - {name: score, type: double}
 filters:
   - type: to_json
+    column:
+      name: test
+      type: string
+    skip_if_null: [id]
 out:
   type: stdout

src/main/java/org/embulk/filter/to_json/ColumnVisitorToJsonImpl.java

@@ -0,0 +1,240 @@
+package org.embulk.filter.to_json;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.embulk.config.ConfigException;
+import org.embulk.spi.Column;
+import org.embulk.spi.ColumnVisitor;
+import org.embulk.spi.PageBuilder;
+import org.embulk.spi.PageReader;
+import org.embulk.spi.Schema;
+import org.embulk.spi.json.JsonParser;
+import org.embulk.spi.type.Types;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by takahiro.nakayama on 1/23/16.
+ */
+public class ColumnVisitorToJsonImpl
+        implements ColumnVisitor
+{
+    private static class ColumnSetter
+    {
+        private final PageBuilder pageBuilder;
+        private final JsonStringSetter jsonStringSetter;
+        private final JsonParser jsonParser = new JsonParser();
+
+        interface JsonStringSetter
+        {
+            void set(String json);
+
+            void set(Column column, String json);
+
+            void set(int index, String json);
+        }
+
+        public ColumnSetter(final PageBuilder pageBuilder, final Column outputColumn)
+        {
+            this.pageBuilder = pageBuilder;
+
+            if (Types.STRING.equals(outputColumn.getType())) {
+                this.jsonStringSetter = new JsonStringSetter()
+                {
+                    @Override
+                    public void set(String json)
+                    {
+                        set(outputColumn, json);
+                    }
+
+                    @Override
+                    public void set(Column column, String json)
+                    {
+                        set(column.getIndex(), json);
+                    }
+
+                    @Override
+                    public void set(int index, String json)
+                    {
+                        setAsString(index, json);
+                    }
+
+                    private void setAsString(int index, String json)
+                    {
+                        pageBuilder.setString(index, json);
+                    }
+                };
+            }
+            else if (Types.JSON.equals(outputColumn.getType())) {
+                this.jsonStringSetter = new JsonStringSetter()
+                {
+                    @Override
+                    public void set(String json)
+                    {
+                        set(outputColumn, json);
+                    }
+
+                    @Override
+                    public void set(Column column, String json)
+                    {
+                        set(column.getIndex(), json);
+                    }
+
+                    @Override
+                    public void set(int index, String json)
+                    {
+                        setAsJson(index, json);
+                    }
+
+                    private void setAsJson(int index, String json)
+                    {
+                        pageBuilder.setJson(index, jsonParser.parse(json));
+                    }
+                };
+            }
+            else {
+                throw new ConfigException(String.format("Cannot convert JSON to type: %s", outputColumn.getType()));
+            }
+        }
+    }
+
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+    private final Map<String, Object> map;
+    private final PageReader pageReader;
+    private final ColumnSetter columnSetter;
+    private List<String> skipColumnsIfNull = Lists.newArrayList();
+    private boolean skipRecordFlag = false;
+
+    ColumnVisitorToJsonImpl(PageReader pageReader, PageBuilder pageBuilder,
+            Column outputColumn, List<String> skipColumnsIfNull)
+    {
+        this.map = Maps.newHashMap();
+        this.pageReader = pageReader;
+        this.columnSetter = new ColumnSetter(pageBuilder, outputColumn);
+        this.skipColumnsIfNull = skipColumnsIfNull;
+    }
+
+    @Override
+    public void booleanColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+        map.put(column.getName(), pageReader.getBoolean(column));
+    }
+
+    @Override
+    public void longColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+        map.put(column.getName(), pageReader.getLong(column));
+    }
+
+    @Override
+    public void doubleColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+        map.put(column.getName(), pageReader.getDouble(column));
+    }
+
+    @Override
+    public void stringColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+        map.put(column.getName(), pageReader.getString(column));
+    }
+
+    @Override
+    public void timestampColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+        map.put(column.getName(), pageReader.getTimestamp(column).toString());
+    }
+
+    @Override
+    public void jsonColumn(Column column)
+    {
+        if (pageReader.isNull(column)) {
+            setSkipRecordFlagIfNeeded(column);
+            putNull(column);
+            return;
+        }
+
+        try {
+            map.put(column.getName(), objectMapper.readTree(pageReader.getJson(column).toJson()));
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void putNull(Column column)
+    {
+        map.put(column.getName(), null);
+    }
+
+    private void clear()
+    {
+        skipRecordFlag = false;
+        map.clear();
+    }
+
+    private void setSkipRecordFlagIfNeeded(Column column)
+    {
+        if (skipColumnsIfNull.contains(column.getName())) {
+            skipRecordFlag = true;
+        }
+    }
+
+    private boolean shouldSkipRecord()
+    {
+        return skipRecordFlag;
+    }
+
+    public void visit()
+    {
+        pageReader.getSchema().visitColumns(this);
+
+        try {
+            if (!shouldSkipRecord()) {
+                columnSetter.jsonStringSetter.set(buildJsonString());
+            }
+        }
+        finally {
+            clear();
+        }
+    }
+
+    private String buildJsonString()
+    {
+        try {
+            return objectMapper.writeValueAsString(map);
+        }
+        catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}