import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
HashMap<Column, Map<String, Integer>> getColumnOptions(PluginTask task)
{
    SchemaConfig schemaConfig = task.getSchema();
    Schema schema = schemaConfig.toSchema();
    HashMap<Column, Map<String, Integer>> lengthMap = new HashMap<>();
    for (Column column : schema.getColumns()) {
        HashMap<String, Integer> miniMap = new HashMap<>();
        ColumnConfig c = schemaConfig.getColumn(column.getIndex());
        miniMap.put("length", c.getOption().get(Integer.class, "length", 0));
        miniMap.put("null_rate", c.getOption().get(Integer.class, "null_rate", 0));
        miniMap.put("max_value", c.getOption().get(Integer.class, "max_value", null));
        miniMap.put("min_value", c.getOption().get(Integer.class, "min_value", null));
        lengthMap.put(column, miniMap);
    }
    return lengthMap;
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
public KafkaInputColumns(KafkaInputPlugin.PluginTask task) {

        List<String> columns = task.getColumns();

        // Validate column name
        if(columns.size() <= 0) {
            throw new IllegalArgumentException("columns property should have one or more elements.");
        }
        for(String column : columns) {
            if(column == null) {
                throw new IllegalArgumentException("column name should not be null.");
            }
            if(!ArrayUtils.contains(columnList, column.toLowerCase())) {
                throw new IllegalArgumentException("column: \"" + column + "\" does not use in columns property.");
            }
        }

        columnHandler = new ColumnBase[columns.size()];
        List<ColumnConfig> columnConfigs = new LinkedList<>();
        for(int i = 0; i < columnHandler.length; i++) {
            columnHandler[i] = getHandler(columns.get(i), task);
            columnConfigs.add(columnHandler[i].getColumnConfig());
        }

        this.schema = new SchemaConfig(columnConfigs).toSchema();
    }
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public void transaction(ConfigSource configSource, Control control)
{
    PluginTask task = configSource.loadConfig(PluginTask.class);

    if (! task.getColumnOptions().isEmpty()) {
        log.warn("embulk-parser-jsonl: \"column_options\" option is deprecated, specify type directly to \"columns\" option with typecast: true (default: true).");
    }

    SchemaConfig schemaConfig = getSchemaConfig(task);
    ImmutableList.Builder<Column> columns = ImmutableList.builder();
    for (int i = 0; i < schemaConfig.getColumnCount(); i++) {
        ColumnConfig columnConfig = schemaConfig.getColumn(i);
        Type type = getType(task, columnConfig);
        columns.add(new Column(i, columnConfig.getName(), type));
    }
    control.run(task.dump(), new Schema(columns.build()));
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private List<ExpandedColumn> initializeExpandedColumns(PluginTask task,
                                                       Schema outputSchema)
{
    ImmutableList.Builder<ExpandedColumn> expandedJsonColumnsBuilder = ImmutableList.builder();
    for (Column outputColumn : outputSchema.getColumns()) {
        for (ColumnConfig expandedColumnConfig : task.getExpandedColumns()) {
            if (outputColumn.getName().equals(expandedColumnConfig.getName())) {

                TimestampParser timestampParser = null;
                if (Types.TIMESTAMP.equals(expandedColumnConfig.getType())) {
                    timestampParser = createTimestampParser(task, expandedColumnConfig);
                }

                ExpandedColumn expandedColumn = new ExpandedColumn(outputColumn.getName(),
                                                                   outputColumn,
                                                                   task.getRoot() + outputColumn.getName(),
                                                                   Optional.fromNullable(timestampParser));
                expandedJsonColumnsBuilder.add(expandedColumn);
            }
        }
    }
    return expandedJsonColumnsBuilder.build();
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private void initializeColumnBean(PluginTask task, Schema schema) {
	List<ColumnConfig> list = task.getColumns().getColumns();

	Map<String, ColumnOptionTask> map = new HashMap<>();
	List<SheetCommonOptionTask> slist = getSheetOption();
	for (int i = slist.size() - 1; i >= 0; i--) {
		SheetCommonOptionTask s = slist.get(i);
		if (s instanceof SheetOptionTask) {
			Optional<Map<String, ColumnOptionTask>> option = ((SheetOptionTask) s).getColumns();
			if (option.isPresent()) {
				map.putAll(option.get());
			}
		}
	}

	for (Column column : schema.getColumns()) {
		String name = column.getName();
		ColumnConfig c = list.get(column.getIndex());
		ColumnOptionTask t = c.getOption().loadConfig(ColumnOptionTask.class);
		PoiExcelColumnBean bean = new PoiExcelColumnBean(this, column, t, map.get(name));
		columnBeanList.add(bean);
	}

	new PoiExcelColumnIndex().initializeColumnIndex(task, columnBeanList);
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
public HashMap<String, HashMap<String, String>> build()
        throws IOException
{
    HashMap<String, HashMap<String, String>> table = Maps.newHashMap();

    for (HashMap<String, String> rawRecord: loadFile()) {

        HashMap<String, String> record = Maps.newHashMap();

        for (ColumnConfig column: columns) {
            String columnKey = columnPrefix + column.getName();
            String value = rawRecord.get(column.getName());

            record.put(columnKey, value);
        }

        String rowKey = rawRecord.get(rowKeyName);
        table.put(rowKey, record);
    }

    return table;
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private Schema buildOutputSchema(Schema inputSchema, List<ColumnConfig> columns, String joinedColumnPrefix)
{
    ImmutableList.Builder<Column> builder = ImmutableList.builder();

    int i = 0; // columns index
    for (Column inputColumn: inputSchema.getColumns()) {
        Column outputColumn = new Column(i++, inputColumn.getName(), inputColumn.getType());
        builder.add(outputColumn);
    }
    for (ColumnConfig columnConfig: columns) {
        String columnName = joinedColumnPrefix + columnConfig.getName();
        builder.add(new Column(i++, columnName, columnConfig.getType()));
    }

    return new Schema(builder.build());
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private HashMap<String, TimestampParser> buildTimestampParserMap(ScriptingContainer jruby, List<ColumnConfig> columns, String joinedColumnPrefix, String timeZone)
{
    final HashMap<String, TimestampParser> timestampParserMap = Maps.newHashMap();
    for (ColumnConfig columnConfig: columns) {
        if (Types.TIMESTAMP.equals(columnConfig.getType())) {
            String format = columnConfig.getOption().get(String.class, "format");
            DateTimeZone timezone = DateTimeZone.forID(timeZone);
            TimestampParser parser = new TimestampParser(jruby, format, timezone);

            String columnName = joinedColumnPrefix + columnConfig.getName();

            timestampParserMap.put(columnName, parser);
        }
    }

    return timestampParserMap;
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private Map<Column, String> createJsonPathMap(PluginTask task, Schema schema)
{
    ImmutableMap.Builder<Column, String> builder = ImmutableMap.builder();
    for (int i = 0; i < schema.size(); i++) {
        ColumnConfig config = getSchemaConfig(task).getColumn(i);
        JsonpathColumnOption option = config.getOption().loadConfig(JsonpathColumnOption.class);
        if (option.getPath().isPresent()) {
            builder.put(schema.getColumn(i), option.getPath().get());
        }
    }
    return builder.build();
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private void validateExpandedColumns(List<ColumnConfig> expandedColumns)
{
    List<String> columnList = new ArrayList<>();
    for (ColumnConfig columnConfig: expandedColumns) {
        String columnName = columnConfig.getName();
        if (columnList.contains(columnName)) {
            throw new ConfigException(String.format("Column config for '%s' is duplicated at 'expanded_columns' option", columnName));
        }
        columnList.add(columnName);
    }
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
public static List<DateParser> create(GrokParserPlugin.PluginTask task) {
    switch (task.getTimestampParser().toLowerCase()) {
        case "ruby":
            TimestampParser[] ps = Timestamps.newTimestampColumnParsers(task, task.getColumns());
            return Arrays.stream(ps)
                    .map(parser -> (DateParser) (text) -> parser.parse(text))
                    .collect(Collectors.toList());
        case "epoch":
            return task.getColumns().getColumns().stream()
                    .map(x -> (DateParser) (text) -> Timestamp.ofEpochMilli(Long.parseLong(text)))
                    .collect(Collectors.toList());
        case "sdf":
        case "simpledateformat":
        default:
            SimpleDateFormat[] parsers = new SimpleDateFormat[task.getColumns().getColumnCount()];
            int i = 0;
            for (ColumnConfig column : task.getColumns().getColumns()) {
                if (column.getType() instanceof TimestampType) {
                    TimestampColumnOption option = column.getOption().loadConfig(TimestampColumnOption.class);
                    String format = convertToJavaDateFormat(option.getFormat().or("yyyy-MM-dd HH:MM:ss.SSS z"));
                    SimpleDateFormat sdf = new SimpleDateFormat(format, Locale.ENGLISH);
                    sdf.setTimeZone(option.getTimeZone().or(DateTimeZone.UTC).toTimeZone());
                    parsers[i] = sdf;
                }
                i++;
            }
            return Arrays.stream(parsers).map(parser ->
                    (DateParser) (String date) -> {
                        try {
                            return Timestamp.ofEpochMilli(parser.parse(date).getTime());
                        } catch (ParseException e) {
                            throw new GrokRecordValidateException(e);
                        }
                    }).collect(Collectors.toList());
    }
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
public TableBuilder(String filePath, String fileFormat, List<ColumnConfig> columns, String rowKeyName, String columnPrefix)
{
    this.filePath = filePath;
    this.fileFormat = fileFormat;
    this.columns = columns;
    this.rowKeyName = rowKeyName;
    this.columnPrefix = columnPrefix;
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public void transaction(ConfigSource config, ParserPlugin.Control control)
{
    PluginTask task = config.loadConfig(PluginTask.class);
    ArrayList<ColumnConfig> columns = new ArrayList<ColumnConfig>();
    final String columnName = task.getColumnName();

    columns.add(new ColumnConfig(columnName, STRING ,config));

    Schema schema = new SchemaConfig(columns).toSchema();
    control.run(task.dump(), schema);
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("offset", this.type, "%d");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("timestamp", this.type, "%d");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("key", this.type, "%s");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("topic", Types.STRING, "%s");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("partition", this.type, "%d");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
@Override
public ColumnConfig getColumnConfig() {
    return new ColumnConfig("value", this.type, "%s");
}
 

import org.embulk.spi.ColumnConfig; //导入依赖的package包/类
private SchemaConfig schema(ColumnConfig... columns)
{
    return new SchemaConfig(Lists.newArrayList(columns));
}