import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Override
public void configure(Config config) {
  LOG.trace("Configuring Morphline Deriver");

  // Designate which dependency step to act upon
  if (!config.hasPath(STEP_NAME_CONFIG) || config.getString(STEP_NAME_CONFIG).trim().isEmpty()) {
    throw new RuntimeException("Missing parameter, " + STEP_NAME_CONFIG);
  } else {
    this.stepName = config.getString(STEP_NAME_CONFIG);
  }

  // Set up the Morphline configuration, the file must be located on the local file system
  this.morphlineFile = config.getString(MORPHLINE);
  this.morphlineId = config.getString(MORPHLINE_ID);

  if (this.morphlineFile == null || this.morphlineFile.trim().length() == 0) {
    throw new MorphlineCompilationException("Missing or empty Morphline File configuration parameter", null);
  }

  // Construct the StructType schema for the Rows
  List<String> fieldNames = config.getStringList(FIELD_NAMES);
  List<String> fieldTypes = config.getStringList(FIELD_TYPES);
  this.schema = RowUtils.structTypeFor(fieldNames, fieldTypes);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void morphlineCompilationError(
    final @Mocked Compiler compiler
    ) throws Exception {

  new Expectations() {{
    config.getString(MorphlineTranslator.MORPHLINE); result = getResourcePath(MORPHLINE_FILE);
    config.getStringList(MorphlineTranslator.FIELD_NAMES); result = Lists.newArrayList("bar");
    config.getStringList(MorphlineTranslator.FIELD_TYPES); result = Lists.newArrayList("int");

    compiler.compile((File) any, anyString, (MorphlineContext) any, (Command) any); result = new Exception("Compilation exception");
  }};

  stringMorphline.configure(config);
  stringMorphline.translate("The Key", "The Message");
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
public TokenizeText(CommandBuilder builder, Config config, Command parent, Command child, MorphlineContext context) {
  super(builder, config, parent, child, context);
  this.inputFieldName = getConfigs().getString(config, "inputField");
  this.outputFieldName = getConfigs().getString(config, "outputField");      
  String solrFieldType = getConfigs().getString(config, "solrFieldType");      
  Config solrLocatorConfig = getConfigs().getConfig(config, "solrLocator");
  SolrLocator locator = new SolrLocator(solrLocatorConfig, context);
  LOG.debug("solrLocator: {}", locator);
  IndexSchema schema = locator.getIndexSchema();
  FieldType fieldType = schema.getFieldTypeByName(solrFieldType);
  if (fieldType == null) {
    throw new MorphlineCompilationException("Missing Solr field type in schema.xml for name: " + solrFieldType, config);
  }
  this.analyzer = fieldType.getIndexAnalyzer();
  Preconditions.checkNotNull(analyzer);
  try { // register CharTermAttribute for later (implicit) reuse
    this.token = analyzer.tokenStream("content", reader).addAttribute(CharTermAttribute.class);
  } catch (IOException e) {
    throw new MorphlineCompilationException("Cannot create token stream", config, e);
  }
  Preconditions.checkNotNull(token);
  validateArguments();
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
public TokenizeText(CommandBuilder builder, Config config, Command parent, Command child, MorphlineContext context) {
  super(builder, config, parent, child, context);
  this.inputFieldName = getConfigs().getString(config, "inputField");
  this.outputFieldName = getConfigs().getString(config, "outputField");      
  String solrFieldType = getConfigs().getString(config, "solrFieldType");      
  Config solrLocatorConfig = getConfigs().getConfig(config, "solrLocator");
  SolrLocator locator = new SolrLocator(solrLocatorConfig, context);
  LOG.debug("solrLocator: {}", locator);
  IndexSchema schema = locator.getIndexSchema();
  FieldType fieldType = schema.getFieldTypeByName(solrFieldType);
  if (fieldType == null) {
    throw new MorphlineCompilationException("Missing Solr field type in schema.xml for name: " + solrFieldType, config);
  }
  this.analyzer = fieldType.getAnalyzer();
  Preconditions.checkNotNull(analyzer);
  try { // register CharTermAttribute for later (implicit) reuse
    this.token = analyzer.tokenStream("content", reader).addAttribute(CharTermAttribute.class);
  } catch (IOException e) {
    throw new MorphlineCompilationException("Cannot create token stream", config, e);
  }
  Preconditions.checkNotNull(token);
  validateArguments();
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Override
public void configure(Context context) {
  String morphlineFile = context.getString(MORPHLINE_FILE_PARAM);
  String morphlineId = context.getString(MORPHLINE_ID_PARAM);
  if (morphlineFile == null || morphlineFile.trim().length() == 0) {
    throw new MorphlineCompilationException("Missing parameter: " + MORPHLINE_FILE_PARAM, null);
  }
  morphlineFileAndId = morphlineFile + "@" + morphlineId;
  
  if (morphlineContext == null) {
    FaultTolerance faultTolerance = new FaultTolerance(
        context.getBoolean(FaultTolerance.IS_PRODUCTION_MODE, false), 
        context.getBoolean(FaultTolerance.IS_IGNORING_RECOVERABLE_EXCEPTIONS, false),
        context.getString(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES));
    
    morphlineContext = new MorphlineContext.Builder()
      .setExceptionHandler(faultTolerance)
      .setMetricRegistry(SharedMetricRegistries.getOrCreate(morphlineFileAndId))
      .build();
  }
  
  Config override = ConfigFactory.parseMap(
      context.getSubProperties(MORPHLINE_VARIABLE_PARAM + "."));
  morphline = new Compiler().compile(
      new File(morphlineFile), morphlineId, morphlineContext, finalChild, override);
  
  this.mappingTimer = morphlineContext.getMetricRegistry().timer(
      MetricRegistry.name("morphline.app", Metrics.ELAPSED_TIME));
  this.numRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", Metrics.NUM_RECORDS));
  this.numFailedRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numFailedRecords"));
  this.numExceptionRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numExceptionRecords"));
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
/**
 *
 * @param morphlineFile
 * @param morphlineId
 * @param collector
 * @param isProduction
 * @return
 */
public static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction) {
  LOG.debug("Constructing Pipeline[{}#{}]", morphlineFile, morphlineId);

  // Set up the Morphline context and handler
  MorphlineContext context = new MorphlineContext.Builder()
      .setExceptionHandler(new FaultTolerance(isProduction, false))
      .build();

  // Compile the Morphline process
  Command morphline;
  try {
    morphline = new Compiler().compile(
        new File(morphlineFile),
        morphlineId,
        context,
        collector);
  } catch (Exception e) {
    throw new MorphlineCompilationException("Morphline compilation error", null, e);
  }

  // Create the pipeline wrapper
  Pipeline pipeline = new Pipeline(morphline, collector);

  // Ensure shutdown notification to Morphline commands esp in streaming environments
  JVMUtils.closeAtShutdown(pipeline);

  // Prep the pipeline
  Notifications.notifyBeginTransaction(pipeline.getMorphline());

  // Register the pipeline into the cache
  if (null == pipelineCache.get()) {
    pipelineCache.set(new HashMap<String, Pipeline>());
  }
  pipelineCache.get().put(morphlineFile + SEPARATOR + morphlineId, pipeline);

  LOG.trace("Pipeline[{}#{}] prepared", morphlineFile, morphlineId);
  return pipeline;
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Override
public void configure(Config config) {
  LOG.trace("Configuring Morphline Translator");

  // Define the encoding values, if necessary
  this.keyEncoding = config.getString(ENCODING_KEY);
  this.messageEncoding = config.getString(ENCODING_MSG);

  // Set up the Morphline configuration, the file must be located on the local file system
  this.morphlineFile = config.getString(MORPHLINE);
  this.morphlineId = config.getString(MORPHLINE_ID);

  if (this.morphlineFile == null || this.morphlineFile.trim().length() == 0) {
    throw new MorphlineCompilationException("Missing or empty Morphline File configuration parameter", null);
  }

  // Construct the StructType schema for the Rows
  List<String> fieldNames = config.getStringList(FIELD_NAMES);
  List<String> fieldTypes = config.getStringList(FIELD_TYPES);
  this.doesAppendRaw = TranslatorUtils.doesAppendRaw(config);
  if (this.doesAppendRaw) {
    fieldNames.add(TranslatorUtils.getAppendRawKeyFieldName(config));
    fieldTypes.add("binary");
    fieldNames.add(TranslatorUtils.getAppendRawValueFieldName(config));
    fieldTypes.add("binary");
  }
  this.schema = RowUtils.structTypeFor(fieldNames, fieldTypes);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = RuntimeException.class)
public void deriveMorphlineMapperFunctionError(
    final @Mocked MorphlineUtils utils
) throws Exception {

  Map<String, Object> paramMap = new HashMap<>();
  paramMap.put(MorphlineDeriver.STEP_NAME_CONFIG, "dep1");
  paramMap.put(MorphlineDeriver.MORPHLINE, "morphline");
  paramMap.put(MorphlineDeriver.MORPHLINE_ID, "id");
  paramMap.put(MorphlineDeriver.FIELD_NAMES, Lists.newArrayList("bar"));
  paramMap.put(MorphlineDeriver.FIELD_TYPES, Lists.newArrayList("int"));
  final Config config = ConfigFactory.parseMap(paramMap);

  new Expectations() {{
    MorphlineUtils.morphlineMapper(anyString, anyString, (StructType) any); result =
        new MorphlineCompilationException("Compile exception", config);
  }};

  Dataset<Row> dataFrame = Contexts.getSparkSession().createDataFrame(
      Lists.newArrayList(RowFactory.create(1)),
      DataTypes.createStructType(Lists.newArrayList(DataTypes.createStructField("baz", DataTypes.IntegerType, false)))
  );

  Map<String, Dataset<Row>> dependencies = Maps.newHashMap();
  dependencies.put("dep1", dataFrame);

  Deriver deriver = new MorphlineDeriver();
  deriver.configure(config);

  deriver.derive(dependencies);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void emptyMorphlineFile() throws Exception {

  new Expectations() {{
    config.getString(MorphlineTranslator.MORPHLINE); result = "";
  }};

  stringMorphline.configure(config);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void invalidCommand() throws Exception {
  new Expectations() {{
    config.getString(MorphlineTranslator.ENCODING_KEY); result = "UTF-8";
    config.getString(MorphlineTranslator.ENCODING_MSG); result = "UTF-8";
    config.getString(MorphlineTranslator.MORPHLINE); result = getResourcePath(MORPHLINE_FILE);
    config.getString(MorphlineTranslator.MORPHLINE_ID); result = "invalid-command";
    config.getStringList(MorphlineTranslator.FIELD_NAMES); result = Lists.newArrayList("int", "str", "float");
    config.getStringList(MorphlineTranslator.FIELD_TYPES); result = Lists.newArrayList("int", "string", "float");
  }};

  stringMorphline.configure(config);
  stringMorphline.translate("The Key", "The Message");
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
private static SolrContentHandlerFactory getSolrContentHandlerFactory(
    Class<? extends SolrContentHandlerFactory> factoryClass, Collection<String> dateFormats, Config config) {
  try {
    return factoryClass.getConstructor(Collection.class).newInstance(dateFormats);
  } catch (NoSuchMethodException nsme) {
    throw new MorphlineCompilationException("Unable to find valid constructor of type "
      + factoryClass.getName() + " for creating SolrContentHandler", config, nsme);
  } catch (Exception e) {
    throw new MorphlineCompilationException("Unexpected exception when trying to create SolrContentHandlerFactory of type "
      + factoryClass.getName(), config, e);
  }
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
public DocumentLoader getLoader() {
  if (context instanceof SolrMorphlineContext) {
    DocumentLoader loader = ((SolrMorphlineContext)context).getDocumentLoader();
    if (loader != null) {
      return loader;
    }
  }
  
  if (zkHost != null && zkHost.length() > 0) {
    if (collectionName == null || collectionName.length() == 0) {
      throw new MorphlineCompilationException("Parameter 'zkHost' requires that you also pass parameter 'collection'", config);
    }
    CloudSolrServer cloudSolrServer = new CloudSolrServer(zkHost);
    cloudSolrServer.setDefaultCollection(collectionName);
    cloudSolrServer.connect();
    return new SolrServerDocumentLoader(cloudSolrServer, batchSize);
  } else {
    if (solrUrl == null || solrUrl.length() == 0) {
      throw new MorphlineCompilationException("Missing parameter 'solrUrl'", config);
    }
    int solrServerNumThreads = 2;
    int solrServerQueueLength = solrServerNumThreads;
    SolrServer server = new SafeConcurrentUpdateSolrServer(solrUrl, solrServerQueueLength, solrServerNumThreads);
    // SolrServer server = new HttpSolrServer(solrServerUrl);
    // SolrServer server = new ConcurrentUpdateSolrServer(solrServerUrl, solrServerQueueLength, solrServerNumThreads);
    // server.setParser(new XMLResponseParser()); // binary parser is used by default
    return new SolrServerDocumentLoader(server, batchSize);
  }
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
private void validateSchema(IndexSchema schema) {
  if (schema.getUniqueKeyField() == null) {
    throw new MorphlineCompilationException("Solr schema.xml is missing unique key field", config);
  }
  if (!schema.getUniqueKeyField().isRequired()) {
    throw new MorphlineCompilationException("Solr schema.xml must contain a required unique key field", config);
  }
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void setPipelineFileNotFound() throws Exception {
  MorphlineUtils.setPipeline("file", "id", new MorphlineUtils.Collector(), true);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void missingMorphlineFile() throws Exception {
  stringMorphline.configure(config);
}
 

import org.kitesdk.morphline.api.MorphlineCompilationException; //导入依赖的package包/类
public Mapping(Config config, MorphlineContext context) {
    Configs configs = new Configs();
    this.inputColumn = resolveColumnName(configs.getString(config, "inputColumn"));
    this.columnFamily = Bytes.toBytes(splitFamilyAndQualifier(inputColumn)[0]);
    
    String qualifierString = splitFamilyAndQualifier(inputColumn)[1];
    this.isWildCard = qualifierString.endsWith("*");
    if (isWildCard) {
        qualifierString = qualifierString.substring(0, qualifierString.length() - 1);
    }
    this.qualifier = Bytes.toBytes(qualifierString);
    
    String outputField = configs.getString(config, "outputField", null);
    this.outputFieldNames = configs.getStringList(config, "outputFields", null);
    if (outputField == null && outputFieldNames == null) {
      throw new MorphlineCompilationException("Either outputField or outputFields must be defined", config);
    }
    if (outputField != null && outputFieldNames != null) {
      throw new MorphlineCompilationException("Must not define both outputField and outputFields at the same time", config);
    }
    if (outputField == null) {
      this.isDynamicOutputFieldName = false;
      this.outputFieldName = null;
    } else {
      this.isDynamicOutputFieldName = outputField.endsWith("*");
      if (isDynamicOutputFieldName) {
          this.outputFieldName = outputField.substring(0, outputField.length() - 1);
      } else {
          this.outputFieldName = outputField;
      }
    }
    
    this.type = configs.getString(config, "type", "byte[]");
    if (type.equals("byte[]")) { // pass through byte[] to downstream morphline commands without conversion
        this.byteArrayMapper = new ByteArrayValueMapper() {
            @Override
            public Collection map(byte[] input) {
                return Collections.singletonList(input);
            }
        };
    } else {
        this.byteArrayMapper = ByteArrayValueMappers.getMapper(type);
    }

    LowerCaseValueSource source = new Validator<LowerCaseValueSource>().validateEnum(config,
            configs.getString(config, "source", LowerCaseValueSource.value.toString()),
            LowerCaseValueSource.class);

    if (source == LowerCaseValueSource.value) {
        if (isWildCard) {
            this.extractor = new PrefixMatchingCellExtractor(columnFamily, qualifier);
        } else {
            this.extractor = new SingleCellExtractor(columnFamily, qualifier);
        }
    } else {
        if (isWildCard) {
            this.extractor = new PrefixMatchingQualifierExtractor(columnFamily, qualifier);
        } else {
            throw new IllegalArgumentException("Can't create a non-prefix-based qualifier extractor");
        }
    }
    
    configs.validateArguments(config);
    if (context instanceof HBaseMorphlineContext) {
        ((HBaseMorphlineContext)context).getExtractors().add(this.extractor);
    }
}