datacollector:StreamSets Data Collector-连续的大数据和云平台获取基础架构

/* * Copyright 2017 StreamSets Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package com.streamsets.pipeline.stage.processor.fieldtypeconverter; import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.streamsets.pipeline.api.Field; import com.streamsets.pipeline.api.FileRef; import com.streamsets.pipeline.api.OnRecordError; import com.streamsets.pipeline.api.ProtoConfigurableEntity; import com.streamsets.pipeline.api.Record; import com.streamsets.pipeline.api.StageException; import com.streamsets.pipeline.api.base.OnRecordErrorException; import com.streamsets.pipeline.config.DateFormat; import com.streamsets.pipeline.config.DecimalScaleRoundingStrategy; import com.streamsets.pipeline.config.ZonedDateTimeFormat; import com.streamsets.pipeline.sdk.ProcessorRunner; import com.streamsets.pipeline.sdk.RecordCreator; import com.streamsets.pipeline.sdk.StageRunner; import com.streamsets.pipeline.stage.common.HeaderAttributeConstants; import org.junit.Assert; import org.junit.Test; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; import java.math.BigDecimal; import java.nio.charset.Charset; import java.text.SimpleDateFormat; import java.time.ZoneId; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.Arrays; import java.util.Calendar; import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.TimeZone; import static com.streamsets.pipeline.stage.processor.fieldtypeconverter.Errors.CONVERTER_04; import static com.streamsets.testing.Matchers.mapFieldWithEntry; import static org.junit.Assert.assertThat; public class TestFieldTypeConverterProcessorFields { @Test public void testStringToNonExistentField() throws StageException { FieldTypeConverterConfig fieldTypeConverterConfig = new FieldTypeConverterConfig(); fieldTypeConverterConfig.fields = ImmutableList.of("/nonExistent", "/beginner", "/expert", "/skilled"); fieldTypeConverterConfig.targetType = Field.Type.BOOLEAN; fieldTypeConverterConfig.dataLocale = "en"; ProcessorRunner runner = new ProcessorRunner.Builder(FieldTypeConverterDProcessor.class) .addConfiguration("convertBy", ConvertBy.BY_FIELD) .addConfiguration("fieldTypeConverterConfigs", ImmutableList.of(fieldTypeConverterConfig)) .addOutputLane("a").build(); runner.runInit(); try { Map<String, Field> map = new LinkedHashMap<>(); map.put("beginner", Field.create("false")); map.put("intermediate", Field.create("yes")); map.put("advanced", Field.create("no")); map.put("expert", Field.create("true")); map.put("skilled", Field.create("122345566")); map.put("null", Field.create(Field.Type.STRING, null)); Record record = RecordCreator.create("s", "s:1"); record.set(Field.create(map)); StageRunner.Output output = runner.runProcess(ImmutableList.of(record)); Assert.assertEquals(1, output.getRecords().get("a").size()); Field field = output.getRecords().get("a").get(0).get(); Assert.assertTrue(field.getValue() instanceof Map); Map<String, Field> result = field.getValueAsMap(); Assert.assertTrue(result.size() == 6); Assert.assertTrue(result.containsKey("beginner")); Assert.assertEquals(false, result.get("beginner").getValue()); Assert.assertTrue(result.containsKey("intermediate")); Assert.assertEquals("yes", result.get("intermediate").getValue()); Assert.assertTrue(result.containsKey("advanced")); Assert.assertEquals("no", result.get("advanced").getValue()); Assert.assertTrue(result.containsKey("expert")); Assert.assertEquals(true, result.get("expert").getValue()); Assert.assertTrue(result.containsKey("skilled")); Assert.assertEquals(false, result.get("skilled").getValue()); Assert.assertTrue(result.containsKey("null")); Assert.assertEquals(null, result.get("null").getValue()); } finally { runner.runDestroy(); } } @Test public void testStringToBoolean() throws StageException { FieldTypeConverterConfig fieldTypeConverterConfig = new FieldTypeConverterConfig(); fieldTypeConverterConfig.fields = ImmutableList.of("/beginner", "/intermediate", "/skilled", "/advanced", "/expert" , "/null"); fieldTypeConverterConfig.targetType = Field.Type.BOOLEAN; fieldTypeConverterConfig.dataLocale = "en"; ProcessorRunner runner = new ProcessorRunner.Builder(FieldTypeConverterDProcessor.class) .addConfiguration("convertBy", ConvertBy.BY_FIELD) .addConfiguration("fieldTypeConverterConfigs", ImmutableList.of(fieldTypeConverterConfig)) .addOutputLane("a").build(); runner.runInit(); try { Map<String, Field> map = new LinkedHashMap<>(); map.put("beginner", Field.create("false")); map.put("intermediate", Field.create("yes")); map.put("advanced", Field.create("no")); map.put("expert", Field.create("true")); map.put("skilled", Field.create("122345566")); map.put("null", Field.create(Field.Type.STRING, null)); Record record = RecordCreator.create("s", "s:1"); record.set(Field.create(map)); StageRunner.Output output = runner.runProcess(ImmutableList.of(record)); Assert.assertEquals(1, output.getRecords().get("a").size()); Field field = output.getRecords().get("a").get(0).get(); Assert.assertTrue(field.getValue() instanceof Map); Map<String, Field> result = field.getValueAsMap(); Assert.assertTrue(result.size() == 6); Assert.assertTrue(result.containsKey("beginner")); Assert.assertEquals(false, result.get("beginner").getValue()); Assert.assertTrue(result.containsKey("intermediate")); Assert.assertEquals(false, result.get("intermediate").getValue()); Assert.assertTrue(result.containsKey("advanced")); Assert.assertEquals(false, result.get("advanced").getValue()); Assert.assertTrue(result.containsKey("expert")); Assert.assertEquals(true, result.get("expert").getValue()); Assert.assertTrue(result.containsKey("skilled")); Assert.assertEquals(false, result.get("skilled").getValue()); Assert.assertTrue(result.containsKey("null")); Assert.assertEquals(null, result.get("null").getValue()); } finally { runner.runDestroy(); } } @Test public void testBooleanToInt() throws StageException { FieldTypeConverterConfig fieldTypeConverterConfig = new FieldTypeConverterConfig(); fieldTypeConverterConfig.fields = ImmutableList.of("/t", "/f"); fieldTypeConverterConfig.targetType = Field.Type.INTEGER; ProcessorRunner runner = new ProcessorRunner.Builder(FieldTypeConverterDProcessor.class) .addConfiguration("convertBy", ConvertBy.BY_FIELD) .addConfiguration("fieldTypeConverterConfigs", ImmutableList.of(fieldTypeConverterConfig)) .addOutputLane("a").build(); runner.runInit(); try { Map<String, Field> map = new LinkedHashMap<>(); map.put("t", Field.create(true)); map.put("f", Field.create(false)); Record record = RecordCreator.create