@Test public void testSchemaValidation() throws Exception { Schema outputSchema = Schema.recordOf( "smallerSchema", Schema.Field.of("x", Schema.of(Schema.Type.INT)), Schema.Field.of("y", Schema.of(Schema.Type.LONG))); JavaScriptTransform.Config config = new JavaScriptTransform.Config( "function transform(input, emitter, context) { emitter.emit({ 'x':input.intField, 'y':input.longField }); }", outputSchema.toString(), null); Schema inputSchema = Schema.recordOf( "biggerSchema", Schema.Field.of("x", Schema.of(Schema.Type.INT)), Schema.Field.of("y", Schema.of(Schema.Type.LONG)), Schema.Field.of("z", Schema.of(Schema.Type.DOUBLE))); MockPipelineConfigurer configurer = new MockPipelineConfigurer(inputSchema, Collections.emptyMap()); new JavaScriptTransform(config).configurePipeline(configurer); Assert.assertEquals(outputSchema, configurer.getOutputSchema()); config = new JavaScriptTransform.Config( "function transform(input, emitter, context) { emitter.emit({ 'x':input.intField, 'y':input.longField }); }", null, null); new JavaScriptTransform(config).configurePipeline(configurer); Assert.assertEquals(inputSchema, configurer.getOutputSchema()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); init(null, pipelineConfigurer.getStageConfigurer().getFailureCollector()); Schema outputSchema = (schema == null) ? pipelineConfigurer.getStageConfigurer().getInputSchema() : schema; pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); }
JavaScriptTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); init(null, pipelineConfigurer.getStageConfigurer().getFailureCollector()); Schema outputSchema = (schema == null) ? pipelineConfigurer.getStageConfigurer().getInputSchema() : schema; pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } }
JavaScriptTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); init(null, pipelineConfigurer.getStageConfigurer().getFailureCollector()); Schema outputSchema = (schema == null) ? pipelineConfigurer.getStageConfigurer().getInputSchema() : schema; pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } JavaScriptTransform(Config config); }
JavaScriptTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); init(null, pipelineConfigurer.getStageConfigurer().getFailureCollector()); Schema outputSchema = (schema == null) ? pipelineConfigurer.getStageConfigurer().getInputSchema() : schema; pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } JavaScriptTransform(Config config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter); }
JavaScriptTransform extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); init(null, pipelineConfigurer.getStageConfigurer().getFailureCollector()); Schema outputSchema = (schema == null) ? pipelineConfigurer.getStageConfigurer().getInputSchema() : schema; pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } JavaScriptTransform(Config config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }