import org.embulk.spi.FileOutput; //导入依赖的package包/类
CommonsCompressArchiveProvider(PluginTask task, FileOutput fileOutput) {
    this.format = ArchiveFormat.toArchiveFormat(task.getFormat());
    this.underlyingFileOutput = fileOutput;
    this.output = new FileOutputOutputStream(fileOutput, task.getBufferAllocator(), FileOutputOutputStream.CloseMode.FLUSH);
    this.entryNamePrefix = task.getPrefix();
    this.baseNum = baseNumSeq.incrementAndGet();
}
 

import org.embulk.spi.FileOutput; //导入依赖的package包/类
OutputStreamFileOutput.Provider createProvider(PluginTask task, FileOutput output) {
    String format = task.getFormat();
    if (CommonsCompressCompressorProvider.isCompressorFormat(format)) {
        return new CommonsCompressCompressorProvider(task, output);
    } else if (CommonsCompressArchiveProvider.isArchiveFormat(format)) {
        return new CommonsCompressArchiveProvider(task, output);
    } else {
        throw new RuntimeException("Unknown format is set format:" + format);
    }
}
 

import org.embulk.spi.FileOutput; //导入依赖的package包/类
@Override
public PageOutput open(TaskSource taskSource, final Schema schema, FileOutput output) {
	final PluginTask task = taskSource.loadTask(PluginTask.class);

	final Sheet sheet = newWorkbook(task);

	final FileOutputOutputStream stream = new FileOutputOutputStream(output, task.getBufferAllocator(),
			CloseMode.CLOSE);
	stream.nextFile();

	return new PageOutput() {
		private final PageReader pageReader = new PageReader(schema);

		@Override
		public void add(Page page) {
			pageReader.setPage(page);
			PoiExcelColumnVisitor visitor = new PoiExcelColumnVisitor(task, schema, sheet, pageReader);
			while (pageReader.nextRecord()) {
				schema.visitColumns(visitor);
				visitor.endRecord();
			}
		}

		@Override
		public void finish() {
			Workbook book = sheet.getWorkbook();
			try (FileOutputOutputStream os = stream) {
				book.write(os);
				os.finish();
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		}

		@Override
		public void close() {
			stream.close();
		}
	};
}
 

import org.embulk.spi.FileOutput; //导入依赖的package包/类
@Override
public PageOutput open(final TaskSource taskSource, final Schema inputSchema,
        final FileOutput output)
{
    final PluginTask task = taskSource.loadTask(PluginTask.class);
    final LineEncoder encoder = new LineEncoder(output, task);
    final String nullString = task.getNullString();
    final Schema outputSchema = getOutputSchema(task.getColumnKeys(), inputSchema);

    encoder.nextFile();

    return new PageOutput() {
        private final PageReader pageReader = new PageReader(inputSchema);

        public void add(Page page)
        {
            pageReader.setPage(page);
            while (pageReader.nextRecord()) {
                List<Object> values = Lists.newArrayList();
                for (Column column : outputSchema.getColumns()) {
                    if (pageReader.isNull(column)) {
                        values.add(nullString);
                        continue;
                    }
                    if (Types.STRING.equals(column.getType())) {
                        values.add(pageReader.getString(column));
                    } else if (Types.BOOLEAN.equals(column.getType())) {
                        values.add(pageReader.getBoolean(column));
                    } else if (Types.DOUBLE.equals(column.getType())) {
                        values.add(pageReader.getDouble(column));
                    } else if (Types.LONG.equals(column.getType())) {
                        values.add(pageReader.getLong(column));
                    } else if (Types.TIMESTAMP.equals(column.getType())) {
                        values.add(pageReader.getTimestamp(column));
                    } else if (Types.JSON.equals(column.getType())) {
                        values.add(pageReader.getJson(column));
                    }
                }
                logger.debug("record => {}", values);
                final String text = String.format(task.getFormat(), (Object[]) values.toArray());
                encoder.addText(text);
            }
        }

        public void finish()
        {
            encoder.finish();
        }

        public void close()
        {
            encoder.close();
        }
    };
}
 

import org.embulk.spi.FileOutput; //导入依赖的package包/类
@Override
public FileOutput open(TaskSource taskSource, final FileOutput fileOutput) {
    PluginTask task = taskSource.loadTask(PluginTask.class);
    return new OutputStreamFileOutput(createProvider(task, fileOutput));
}
 

import org.embulk.spi.FileOutput; //导入依赖的package包/类
CommonsCompressCompressorProvider(PluginTask task, FileOutput fileOutput) {
    this.format = CompressorFormat.toCompressorFormat(task.getFormat());
    this.underlyingFileOutput = fileOutput;
    this.output = new FileOutputOutputStream(fileOutput, task.getBufferAllocator(), FileOutputOutputStream.CloseMode.FLUSH);
}