import org.embulk.spi.util.Timestamps; //导入依赖的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.util.Timestamps; //导入依赖的package包/类
protected final TimestampParser getTimestampParser(Column column) {
	if (timestampParsers == null) {
		PluginTask task = visitorValue.getPluginTask();
		timestampParsers = Timestamps.newTimestampColumnParsers(task, task.getColumns());
	}
	return timestampParsers[column.getIndex()];
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
SpeedControlPageOutput(PluginTask task, Schema schema, PageOutput pageOutput) {
    this.controller = new SpeedometerSpeedController(task, SpeedometerSpeedAggregator.getInstance(task));
    this.schema = schema;
    this.allocator = task.getBufferAllocator();
    this.delimiterLength = task.getDelimiter().length();
    this.recordPaddingSize = task.getRecordPaddingSize();
    this.pageReader = new PageReader(schema);
    this.timestampFormatters = Timestamps.newTimestampColumnFormatters(task, schema, task.getColumnOptions());
    this.pageBuilder = new PageBuilder(allocator, schema, pageOutput);
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
private ParquetWriter<PageReader> createWriter(PluginTask task, Schema schema, int processorIndex)
{
    final TimestampFormatter[] timestampFormatters = Timestamps.newTimestampColumnFormatters(task, schema, task.getColumnOptions());

    final Path path = new Path(buildPath(task, processorIndex));
    final CompressionCodecName codec = CompressionCodecName.valueOf(task.getCompressionCodec());
    final int blockSize = task.getBlockSize();
    final int pageSize = task.getPageSize();
    final Configuration conf = createConfiguration(task.getExtraConfigurations());
    final boolean overwrite = task.getOverwrite();

    ParquetWriter<PageReader> writer = null;
    try {
        EmbulkWriterBuilder builder = new EmbulkWriterBuilder(path, schema, timestampFormatters)
                .withCompressionCodec(codec)
                .withRowGroupSize(blockSize)
                .withPageSize(pageSize)
                .withDictionaryPageSize(pageSize)
                .withConf(conf);

        if (overwrite) {
            builder.withWriteMode(ParquetFileWriter.Mode.OVERWRITE);
        }

        writer = builder.build();
    }
    catch (IOException e) {
        Throwables.propagate(e);
    }
    return writer;
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
ColumnVisitorImpl(Map<String, String> row, PluginTask task, PageBuilder pageBuilder) {
    this.row = row;
    this.pageBuilder = pageBuilder;

    this.timestampParsers = Timestamps.newTimestampColumnParsers(
            task, task.getColumns());
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
@Override
public void run(TaskSource taskSource, Schema schema, FileInput input, PageOutput output)
{
    PluginTask task = taskSource.loadTask(PluginTask.class);

    setColumnNameValues(schema);

    final SchemaConfig schemaConfig = getSchemaConfig(task);
    final TimestampParser[] timestampParsers = Timestamps.newTimestampColumnParsers(task, schemaConfig);
    final LineDecoder decoder = newLineDecoder(input, task);
    final JsonParser jsonParser = newJsonParser();
    final boolean stopOnInvalidRecord = task.getStopOnInvalidRecord();

    try (final PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), schema, output)) {
        ColumnVisitorImpl visitor = new ColumnVisitorImpl(task, schema, pageBuilder, timestampParsers);

        while (decoder.nextFile()) { // TODO this implementation should be improved with new JsonParser API on Embulk v0.8.3
            lineNumber = 0;

            while ((line = decoder.poll()) != null) {
                lineNumber++;

                try {
                    Value value = jsonParser.parse(line);

                    if (!value.isMapValue()) {
                        throw new JsonRecordValidateException("Json string is not representing map value.");
                    }

                    final Map<Value, Value> record = value.asMapValue().map();
                    for (Column column : schema.getColumns()) {
                        Value v = record.get(getColumnNameValue(column));
                        visitor.setValue(v);
                        column.visit(visitor);
                    }

                    pageBuilder.addRecord();
                }
                catch (JsonRecordValidateException | JsonParseException e) {
                    if (stopOnInvalidRecord) {
                        throw new DataException(String.format("Invalid record at line %d: %s", lineNumber, line), e);
                    }
                    log.warn(String.format("Skipped line %d (%s): %s", lineNumber, e.getMessage(), line));
                }
            }
        }

        pageBuilder.finish();
    }
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
protected final TimestampFormatter getTimestampFormatter(Column column) {
	if (timestampFormatters == null) {
		timestampFormatters = Timestamps.newTimestampColumnFormatters(task, schema, task.getColumnOptions());
	}
	return timestampFormatters[column.getIndex()];
}
 

import org.embulk.spi.util.Timestamps; //导入依赖的package包/类
@Override
public void run(TaskSource taskSource, Schema schema, FileInput input, PageOutput output) {
    PluginTask task = taskSource.loadTask(PluginTask.class);
    LineDecoder lineDecoder = new LineDecoder(input, task);
    PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), schema, output);
    TimestampParser[] timestampParsers = Timestamps.newTimestampColumnParsers(task, task.getSchemaConfig());

    Pattern pattern = Pattern.compile(task.getRegex());
    Map<String, DynamicColumnSetter> setterMap = setupSetters(pageBuilder, task.getSchemaConfig(),
            timestampParsers, taskSource.loadTask(PluginTaskFormatter.class));

    while (input.nextFile()) {

        int lineNumber = 0;

        while (true) {
            String line = lineDecoder.poll();
            lineNumber++;

            if (line == null) {
                break;
            }
            Matcher matcher = pattern.matcher(line);
            if (!matcher.matches()) {
                if (task.getSkipIfUnmatch()) {
                    log.warn(String.format("Skipped unmatched line %d: %s", lineNumber, line));
                    continue;
                } else {
                    throw new DataException(String.format("Unmatched Line at line %d: %s", lineNumber, line));
                }
            }

            for (Map.Entry<String, DynamicColumnSetter> pair : setterMap.entrySet()) {
                String value = matcher.group(pair.getKey());
                if (value == null) {
                    pair.getValue().setNull();
                } else {
                    pair.getValue().set(value);
                }
            }
            pageBuilder.addRecord();
        }
    }
    pageBuilder.finish();
}