frsyuki

List<String> columnNames = task.getColumns();
ImmutableList.Builder<Column> builder = ImmutableList.builder();
int i = 0;
for (String columnName : columnNames) {
    for (Column inputColumn: inputSchema.getColumns()) {
        if (inputColumn.getName().equals(columnName)) {
            Column outputColumn = new Column(i++, columnName, inputColumn.getType());
            builder.add(outputColumn);
            break;
        }

frsyuki

class Relation(Model):  # Jsonable.Model
    engine_name = json_attr(str)
    plan = json_attr(Plan)

class ValuesPlan(Plan):
    schema = json_attr(Schema)
    values = json_attr([Literal])  # list of Literal
    alias_name = json_attr(str, optional=True)  # optional

class Plan(Model):

frsyuki

--- ./hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ApplicationClassLoader.java	2015-04-06 15:11:21.000000000 -0700
+++ ./hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ApplicationClassLoader.java.load	2015-04-06 15:11:35.000000000 -0700
@@ -116,6 +116,14 @@
       if (LOG.isDebugEnabled()) {
         LOG.debug("getResource("+name+")=" + url);
       }
+    } else if (!isSystemClass(name, systemClasses)) {
+      url= findResource(name);
+      if (url == null && name.startsWith("/")) {
+        if (LOG.isDebugEnabled()) {

frsyuki

frsyuki

module Embulk
  module Parser

    class XxxParserPlugin < ParserPlugin
      Plugin.register_parser("xxx", self)

      def self.transaction(config, &control)
        parser_task = config.load_config(Java::LineDecoder::DecoderTask)

        # configuration code:

frsyuki

package org.embulk.input;

import java.util.List;
import java.util.ArrayList;
import com.google.common.base.Optional;
import org.embulk.config.CommitReport;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.config.ConfigDiff;
import org.embulk.config.ConfigSource;

frsyuki

reqirb(main):001:0> require 'time'
=> true
irb(main):002:0> Time.strptime("2015-02-24 21:02:57.321", "%Y-%m-%d %H:%m:%S.%N").strftime("%Y-%m-%d %H:%m:%S.%N")
=> "2015-02-24 21:02:57.321000000"
irb(main):003:0> Time.strptime("1424842137.321", "%s.%N").strftime("%Y-%m-%d %H:%m:%S.%N")
=> "2015-02-24 21:02:57.000000000"
irb(main):004:0> RUBY_VERSION
=> "2.1.3"

frsyuki

def self.transaction(config, &control)
  task = {
    'charset' => 'utf-8',
    'newline' => 'CRLF',  # or LF
  }
  # ...
  yield(task)
end

frsyuki

Embulk-plugin-input-pcapng-files書いた にコメント。

本来的にはinputプラグインではなくFile input内のparser/decoderプラグインとして造るべきでは？

その通りで、本来はparserプラグインでしょう。Java側にはParserプラグインを書くAPIがあるのですが、まだそれをJRubyで書く仕組みが無いのです。書きます。 （https://github.com/embulk/embulk/issues/33）

処理対象のファイル一覧→スレッドへの分配あたりを再開発してる感

ファイルを読み込むタイプのプラグインはスレッド数を考えなくても良くて、基本的にファイル数 = タスク数にすればOKです（example_input.rbがまぎらわしいかもしれませんが…）。なぜなら、設定ファイルのexec:のところにスレッド数を書けるからです。

frsyuki