@Test public void testTableSinkWithComplexTypeInOutputSchema() { 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("complex", Schema.mapOf(Schema.of(Schema.Type.STRING), Schema.of(Schema.Type.STRING)) )); 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("complex", Schema.mapOf(Schema.of(Schema.Type.STRING), Schema.of(Schema.Type.STRING)) )); TableSinkConfig tableSinkConfig = new TableSinkConfig("tableSink", "rowkey", outputSchema.toString()); TableSink tableSink = new TableSink(tableSinkConfig); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(inputSchema); tableSink.configurePipeline(mockPipelineConfigurer); FailureCollector collector = mockPipelineConfigurer.getStageConfigurer().getFailureCollector(); Assert.assertEquals(1, collector.getValidationFailures().size()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Preconditions.checkArgument(tableSinkConfig.containsMacro(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD) || !Strings.isNullOrEmpty(tableSinkConfig.getRowField()), "Row field must be given as a property."); Schema outputSchema = SchemaValidator.validateOutputSchemaAndInputSchemaIfPresent(tableSinkConfig.getSchemaStr(), tableSinkConfig.getRowField(), pipelineConfigurer); if (outputSchema != null) { if (outputSchema.getField(tableSinkConfig.getRowField()) == null) { throw new IllegalArgumentException("Output schema should contain the rowkey column."); } if (outputSchema.getFields().size() == 1) { throw new IllegalArgumentException("Output schema should have columns other than rowkey."); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); }
TableSink extends BatchWritableSink<StructuredRecord, byte[], Put> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Preconditions.checkArgument(tableSinkConfig.containsMacro(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD) || !Strings.isNullOrEmpty(tableSinkConfig.getRowField()), "Row field must be given as a property."); Schema outputSchema = SchemaValidator.validateOutputSchemaAndInputSchemaIfPresent(tableSinkConfig.getSchemaStr(), tableSinkConfig.getRowField(), pipelineConfigurer); if (outputSchema != null) { if (outputSchema.getField(tableSinkConfig.getRowField()) == null) { throw new IllegalArgumentException("Output schema should contain the rowkey column."); } if (outputSchema.getFields().size() == 1) { throw new IllegalArgumentException("Output schema should have columns other than rowkey."); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } }
TableSink extends BatchWritableSink<StructuredRecord, byte[], Put> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Preconditions.checkArgument(tableSinkConfig.containsMacro(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD) || !Strings.isNullOrEmpty(tableSinkConfig.getRowField()), "Row field must be given as a property."); Schema outputSchema = SchemaValidator.validateOutputSchemaAndInputSchemaIfPresent(tableSinkConfig.getSchemaStr(), tableSinkConfig.getRowField(), pipelineConfigurer); if (outputSchema != null) { if (outputSchema.getField(tableSinkConfig.getRowField()) == null) { throw new IllegalArgumentException("Output schema should contain the rowkey column."); } if (outputSchema.getFields().size() == 1) { throw new IllegalArgumentException("Output schema should have columns other than rowkey."); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } TableSink(TableSinkConfig tableSinkConfig); }
TableSink extends BatchWritableSink<StructuredRecord, byte[], Put> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Preconditions.checkArgument(tableSinkConfig.containsMacro(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD) || !Strings.isNullOrEmpty(tableSinkConfig.getRowField()), "Row field must be given as a property."); Schema outputSchema = SchemaValidator.validateOutputSchemaAndInputSchemaIfPresent(tableSinkConfig.getSchemaStr(), tableSinkConfig.getRowField(), pipelineConfigurer); if (outputSchema != null) { if (outputSchema.getField(tableSinkConfig.getRowField()) == null) { throw new IllegalArgumentException("Output schema should contain the rowkey column."); } if (outputSchema.getFields().size() == 1) { throw new IllegalArgumentException("Output schema should have columns other than rowkey."); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } 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); }
TableSink extends BatchWritableSink<StructuredRecord, byte[], Put> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); Preconditions.checkArgument(tableSinkConfig.containsMacro(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD) || !Strings.isNullOrEmpty(tableSinkConfig.getRowField()), "Row field must be given as a property."); Schema outputSchema = SchemaValidator.validateOutputSchemaAndInputSchemaIfPresent(tableSinkConfig.getSchemaStr(), tableSinkConfig.getRowField(), pipelineConfigurer); if (outputSchema != null) { if (outputSchema.getField(tableSinkConfig.getRowField()) == null) { throw new IllegalArgumentException("Output schema should contain the rowkey column."); } if (outputSchema.getFields().size() == 1) { throw new IllegalArgumentException("Output schema should have columns other than rowkey."); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } 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);  }