TableSink extends BatchWritableSink<StructuredRecord, byte[], Put> { public TableSink(TableSinkConfig tableSinkConfig) { super(tableSinkConfig); this.tableSinkConfig = tableSinkConfig; } TableSink(TableSinkConfig tableSinkConfig); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(BatchSinkContext context); @Override void initialize(BatchRuntimeContext context); @Override void transform(StructuredRecord input, Emitter<KeyValue<byte[], Put>> emitter);  }
@Test public void testTableSink() { Schema outputSchema = Schema.recordOf( "purchase", Schema.Field.of("rowkey", Schema.of(Schema.Type.STRING)), Schema.Field.of("user", Schema.of(Schema.Type.STRING)), Schema.Field.of("count", Schema.of(Schema.Type.INT)) ); Schema inputSchema = Schema.recordOf( "purchase", Schema.Field.of("rowkey", Schema.of(Schema.Type.STRING)), Schema.Field.of("user", Schema.of(Schema.Type.STRING)), Schema.Field.of("count", Schema.of(Schema.Type.INT)) ); TableSinkConfig tableSinkConfig = new TableSinkConfig("tableSink", "rowkey", outputSchema.toString()); TableSink tableSink = new TableSink(tableSinkConfig); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(inputSchema); tableSink.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(outputSchema, mockPipelineConfigurer.getOutputSchema()); }