2.StreamSets-api.md · fork/streamsets-start-asset_1

##  streamsets-api

├─base 基本Stage抽象实现
│  └─configurablestage 配置化基类
├─credential              证书
├─delegate             委派
│  └─exported
├─el
├─ext
│  ├─io
│  └─json
├─impl
│  └─annotationsprocessor
├─interceptor      拦截
├─lineage           血缘血统
└─service
    ├─dataformats
    │  └─log
    └─sshtunnel

### 主要接口定义全貌

![image-20200314162051872](./image/Analysis-ALL.png)

## javax.annotation.processing.Processor

```java
com.streamsets.pipeline.api.impl.annotationsprocessor.PipelineAnnotationsProcessor
```

```java
@SupportedAnnotationTypes({
    "com.streamsets.pipeline.api.StageDef", 用于声明数据收集器stage类的注释。类必须实现源、处理器或目标。
    "com.streamsets.pipeline.api.GenerateResourceBundle", 注释用于标记阶段和相关类的资源包的生成，这些阶段和相关类具有显示在UI中且需要本地化的标签或消息。
    "com.streamsets.pipeline.api.ElDef", 使用ElDef注释的类会自动扫描EL常量和EL函数定义，它们可用于stage library 的所有配置。
    "com.streamsets.pipeline.api.lineage.LineagePublisherDef",
    "com.streamsets.pipeline.api.credential.CredentialStoreDef",  声明用于自动发现的CredentialStore的实现。
    "com.streamsets.pipeline.api.service.ServiceDef", 用于声明数据收集器服务类的注释。服务必须实现Service，然后在提供属性中指定接口。
    "com.streamsets.pipeline.api.delegate.StageLibraryDelegateDef", 阶段库（Stage library ）委托允许数据收集器框架将某些核心功能“offload”到阶段库中，以隔离这些功能的依赖关系。集群作业提交就是一个很好的例子。
    "com.streamsets.pipeline.api.interceptor.InterceptorDef"
})
```

### datacollector-resource-bundles.json

```json
[
  "com.streamsets.pipeline.api.OnRecordError",  枚举通过内置的“on record error”配置定义阶段的错误记录处理。
  "com.streamsets.pipeline.api.StageDef$DefaultOutputStreams",
  "com.streamsets.pipeline.api.StageDef$VariableOutputStreams",
  "com.streamsets.pipeline.api.ExecutionMode",
  "com.streamsets.pipeline.api.DeliveryGuarantee",
  "com.streamsets.pipeline.api.base.Errors",
  "com.streamsets.pipeline.api.StageUpgrader"
]
```

```java
package com.streamsets.pipeline.api;

/**
 * Enumeration defining the error record handling for stages via the built-in 'on record error' configuration.
 */
// we are using the annotation for reference purposes only.
// the annotation processor does not work on this maven project
// we have a hardcoded 'datacollector-resource-bundles.json' file in resources
@GenerateResourceBundle
public enum OnRecordError implements Label {
  DISCARD("Discard"),
  TO_ERROR("Send to Error"),
  STOP_PIPELINE("Stop Pipeline"),
  ;

private final String label;

OnRecordError(String label) {
    this.label = label;
  }

@Override
  public String getLabel() {
    return label;
  }

}

```

```java
public @interface StageDef {

//enum for processors using LanePredicateMapping configurations  enum用于使用LanePredicateMapping配置的处理器

// we are using the annotation for reference purposes only.  我们使用注释仅作参考之用。
  // the annotation processor does not work on this maven project 注释处理器不能在这个maven项目上工作
  // we have a hardcoded 'datacollector-resource-bundles.json' file in resources 在resources中有一个硬编码的“datacollector-resource-bundle.json”文件

/**
   * Enumeration to use in {@link #outputStreams()} to indicate that the output streams are variable, driven by the
   * stage configuration specified in {@link #outputStreamsDrivenByConfig()}.
   * 在outputStreams()中使用的枚举，以指示输出流是变量，由outputStreamsDrivenByConfig()中指定的stage配置驱动。
   */
  @GenerateResourceBundle
  enum VariableOutputStreams implements Label {
    ;

@Override
    public String getLabel() {
      return null;
    }
  }

/**
   * Default enumeration used by stages having a single output stream.
   */
  @GenerateResourceBundle
  public enum DefaultOutputStreams implements Label {
    OUTPUT("Output");

private final String label;

DefaultOutputStreams(String label) {
      this.label = label;
    }

@Override
    public String getLabel() {
      return label;
    }
  }

```

```java
package com.streamsets.pipeline.api;

/**
 * Defines the execution modes of a pipeline.
 * 
 * It is used in stage definitions to restrict the execution modes the stage supports.
 * 
 * Accessible via the stage context to know the current execution mode of the pipeline.
 * 
定义管道的执行模式。
它在Stage阶段定义中用于限制阶段支持的执行模式。
可通过阶段上下文访问，以了解管道的当前执行模式。
 * @see StageDef#execution()
 * @see Stage.Context#getExecutionMode()
 */
public enum ExecutionMode implements Label {
 STANDALONE("Standalone"),
 @Deprecated
 CLUSTER("Cluster"), //Kept for backward compatibility - replaced by CLUSTER_BATCH and CLUSTER_STREAMING 保留为向后兼容——替换为CLUSTER_BATCH和CLUSTER_STREAMING
 CLUSTER_BATCH("Cluster Batch"),
 CLUSTER_YARN_STREAMING("Cluster Yarn Streaming"),
 CLUSTER_MESOS_STREAMING("Cluster Mesos Streaming"),
 SLAVE("Slave"),
 EDGE("Edge"),
 EMR_BATCH("Cluster EMR Batch"),
 BATCH("Batch"),
 STREAMING("Streaming"),
 ;

private final String label;

ExecutionMode(String label) {
    this.label = label;
  }

/**
   * Returns the default value of localizable label, for the UI, of the execution mode enum.
   *
   * @return the default value of localizable label.
   */
  @Override
  public String getLabel() {
    return label;
  }
}

```

```java
package com.streamsets.pipeline.api;

/**
 * Delivery guarantee of a pipeline.
 * 管道的交付保证。
 * Currently only exposed for PushSource.
 目前仅用于推送源点PushSource。
 */
public enum DeliveryGuarantee implements Label {
  AT_LEAST_ONCE("At Least Once"),
  AT_MOST_ONCE("At Most Once")
  ;

private final String label;

DeliveryGuarantee(String label) {
    this.label = label;
  }

@Override
  public String getLabel() {
    return label;
  }

}

```

```java
package com.streamsets.pipeline.api.base;

import com.streamsets.pipeline.api.ErrorCode;
import com.streamsets.pipeline.api.GenerateResourceBundle;

/**
 * Error codes produced by the Data Collector where there is a stage configuration issue.
 当存在阶段配置问题时，由数据收集器产生的错误代码。
 */
// we are using the annotation for reference purposes only.
// the annotation processor does not work on this maven project
// we have a hardcoded 'datacollector-resource-bundles.json' file in resources
@GenerateResourceBundle
public enum Errors implements ErrorCode {
  API_00("Stage '{}' requires exactly {} output streams. There are '{}'."),
  API_01("Cannot convert {} field '{}' to Boolean"),
  API_02("Cannot convert {} field '{}' to Byte[]"),
  API_03("Cannot convert Byte[] to {}"),
  API_04("Cannot convert {} field '{}' to Byte"),
  API_05("Cannot convert {} field '{}' to Char"),
  API_06("Cannot parse '{}' to a Date. Use the following ISO 8601 UTC date format: yyyy-MM-dd'T'HH:mm'Z'."),
  API_07("Cannot convert {} field '{}' to Date"),
  API_08("Cannot convert {} field '{}' to Decimal"),
  API_09("Cannot convert {} field '{}' to Double"),
  API_10("Cannot convert {} field '{}' to Float"),
  API_11("Cannot convert {} field '{}' to Integer"),
  API_12("Cannot convert {} field '{}' to List"),
  API_13("Cannot convert List to {}"),
  API_14("Cannot convert {} field '{}' to Long"),
  API_15("Cannot convert {} field '{}' to Map"),
  API_16("Cannot convert Map to {}"),
  API_17("Cannot convert {} field '{}' to Short"),
  API_18("Cannot convert Map, List, or Byte[] to String"),

API_19("Error while initializing stage: {}"),
  API_20("The stage implementation overridden the init() but didn't call super.init()"),

API_21("Cannot convert {} field '{}' to ListMap"),
  API_22("Cannot convert ListMap to {}"),
  API_23("Cannot convert {} to File Ref Object"),
  API_24("Cannot convert FileRef Object to {}"),
  API_25("Cannot convert {} field '{}' to ZonedDateTime"),
  API_26("Cannot parse '{}' to a ZonedDateTime. " +
      "Use ISO 8601 format with offset and zone, such as '2011-12-03T10:15:30+01:00[Europe/Paris]'"),

;

private final String msg;

Errors(String msg) {
    this.msg = msg;
  }

@Override
  public String getCode() {
    return name();
  }

@Override
  public String getMessage() {
    return msg;
  }

}
```

```java
package com.streamsets.pipeline.api;

import java.util.List;

/**
 * The <code>StageUpgrader</code> allows stages to upgrade their configuration from previous versions of the stage.
 * 
 * The upgrader is called only when the version of a stage configuration in a pipeline is older than the version of
 * the stage being used in the pipeline (this typically happens immediately after a Data Collector or stage library
 * upgrade).
StageUpgrader允许各个阶段从以前的版本升级它们的配置。
只有当管道中的阶段配置版本比管道中使用的阶段版本更旧时，才会调用升级程序(这通常在数据收集器或阶段库升级之后立即发生)。
 */
public interface StageUpgrader {

/**
   * Error codes used by the upgrader.
   */
  @GenerateResourceBundle
  public enum Error implements ErrorCode {
    UPGRADER_00("Upgrader not implemented for stage '{}:{}' instance '{}'"),
    UPGRADER_01("Cannot upgrade stage '{}:{}' instance '{}' from version '{}' to version '{}'"),
    ;

private final String message;

Error(String message) {
      this.message = message;
    }

@Override
    public String getCode() {
      return name();
    }

@Override
    public String getMessage() {
      return message;
    }
  }

/**
 * Default <code>StageUpgrader</code> implementation. It fails all upgrades.
 */
 public static class Default implements StageUpgrader {

/**
 * This implementation always throws an exception.
 */
 @Override
 public List<Config> upgrade(List<Config> configs, Context context) throws StageException {
 throw new StageException(
 Error.UPGRADER_00,
 context.getLibrary(),
 context.getStageName(),
 context.getStageInstance()
 );
 }
 }

/**
   * Upgrade context with various upgrade metadata.
   */
  public static interface Context {

/**
     * Stage Library name
     */
    public String getLibrary();

/**
     * Stage name (identifies this stage uniquely in SDC)
     */
    public String getStageName();

/**
     * Stage instance name (identifies instance of this stage in particular pipeline).
     */
    public String getStageInstance();

/**
     * Current version of the stored configuration.
     */
    public int getFromVersion();

/**
     * Desired target version after upgrade.
     */
    public int getToVersion();

/**
 * Register given service with provided configuration.
 *
 * @param service Service interface that needs to be registered.
 * @param configs Initial configuration for the service.
 */
 public void registerService(Class service, List<Config> configs);
 }

/**
 * Upgrades the stage configuration from a previous version to current version.
 *
 * @param configs Current configuration that needs to be upgraded.
 * @param context Upgrade context with various metadata.
 * @return The upgraded configuration.
 * @throws StageException if the configurations could not be upgraded.
 * @return
 */
 default public List<Config> upgrade(List<Config> configs, Context context) throws StageException {
 return upgrade(
 context.getLibrary(),
 context.getStageName(),
 context.getStageInstance(),
 context.getFromVersion(),
 context.getToVersion(),
 configs
 );
 }

/**
 * Upgrades the stage configuration from a previous version to current version.
 *
 * @param library stage library name.
 * @param stageName stage name.
 * @param stageInstance stage instance name.
 * @param fromVersion version recorded in the stage configuration to upgrade.
 * @param toVersion version of of the stage library, the version to upgrade the configuration to.
 * @param configs The configurations to upgrade.
 * @return The upgraded configuration.
 * @throws StageException if the configurations could not be upgraded.
 */
 @Deprecated
 default public List<Config> upgrade(
 String library,
 String stageName,
 String stageInstance,
 int fromVersion,
 int toVersion,
 List<Config> configs
 ) throws StageException {
 // Default behavior is no-op
 return configs;
 }

}
```

## pipeline.api

## Batch Record EventRecord

![image-20200311220946136](./image/image-20200311220946136.png)

### Batch

数据收集器处理器(Processor)和目标(Target)阶段接收一个批处理的实例，该实例允许它们访问当前批处理中的记录进行处理。

```java
/**
 * Returns an iterator with all the records in the batch for the current stage.
 * 返回一个迭代器，其中包含当前阶段的批处理中的所有记录。
 * 
 * Every time this method is called it returns a new iterator with all records in the batch.
 * 每次调用此方法时，它都会返回一个新的迭代器，其中包含批处理中的所有记录。
 * @return an iterator with all the records in the batch for the current stage. 返回:包含当前阶段批处理中的所有记录的迭代器。
 */
public Iterator<Record> getRecords();
```

### Record

A <code>Record</code> represents the unit of data Data Collector pipelines process. A record has a {@link Field}
representing the data as well as a header for metadata information.

一条<code>记录</code>表示数据收集器流水线处理的数据单元。记录有一个 **Field**

表示数据以及元数据信息的标头。

The **Header** of a <code>Record</code> is the associated metadata, some of which is generated by the Pipeline
runtime container.

<code>记录</code>的**Header**是相关的元数据，其中一些是由管道生成的

运行时容器。

The <code>Record</code> API supports a field-path expressions (a simplified
<a href="http://en.wikipedia.org/wiki/XPath">XPath</a> like syntax) to directly access values within a
<code>Field</code> value data structure:

Record API支持字段路径表达式(简化的类似XPath的语法)来直接访问字段值数据结构中的值:

<ul>
 <li>An empty string <code>""</code> refers to the root element of the <code>Record</code></li>
 <li>A <code>/{NAME}NAME</code> string refers to the '{NAME}' entry in a <code>Map</code> or a
 <code>LIST_MAP</code></li>
 <li>A <code>[{INDEX}]</code>, with {INDEX} being a zero/positive integer, refers to the '{INDEX}' position in a
 <code>List</code> or a <code>LIST_MAP</code></li>
</ul>
A field-path expression can be used to reference a composite data structure of any depth and composition, for

**field-path表达式**可以用来引用一个复合数据结构的深度和结构
example:

<ul>
 <li>/contactInfo/firstName</li>
 <li>/contactInfo/lastName</li>
 <li>/contactInfo/email[[0]]</li>
 <li>/contactInfo/address/firstLine</li>
 <li>/contactInfo/address/secondLine</li>
 <li>/contactInfo/address/city</li>
 <li>/contactInfo/address/state</li>
 <li>/contactInfo/address/zip</li>
 <li>/contactInfo/phone[[0]]/number</li>
 <li>/contactInfo/phone[[0]]/type</li>
 <li>/contactInfo/phone[[1]]/number</li>
 <li>/contactInfo/phone[[1]]/type</li>
</ul>
Using **field-path expressions** is possible to check for existence, access, modify and delete a <code>Field</code> data
structure.

The {@link #get(String)}, {@link #has(String)}, {@link #delete(String)} and {@link #getEscapedFieldPaths()}
methods work using field- path expressions.

使用字段路径表达式可以检查是否存在、访问、修改和删除<code>字段</code>数据结构。

{@link #get(String)}，

{@link #has(String)}，

{@link #delete(String)}

{@link #getEscapedFieldPaths()}

方法使用字段路径表达式工作。

IMPORTANT: Map key names that are not a word or use any of the following 3 special
characters '<code>/</code>', '<code>[</code>' or '<code>]</code>' must be single-quoted or double-quoted, for
example: '<code>'foo bar'/"xyz]"</code>.

重要:Map key不是一个单词或使用以下3个特殊的

字符'<code>/</code>'， '<code>[</code>'或'<code>]</code>'必须是单引号或双引号，因为

例子:“<代码>“foo bar”/“xyz”> < /代码。

![image-20200311234129548](./image/image-20200311234129548.png)

```java
public interface Source extends ProtoSource<Source.Context> {
public interface Processor extends Stage<Processor.Context> {
public interface Target extends Stage<Target.Context> {
```

### BatchContext

![image-20200312075921536](./image/image-20200312075921536.png)

### ProtoConfigurableEntity

典型配置实体，为所有可配置的对象，包括Stage、Service提供一个共同的基础接口。

```java
public interface ELContext {}
public interface Context extends ConfigIssueContext, ELContext, MetricContext {
 // 从data collector 主配置文件中获取Stage相关的配置信息
 public Configuration getConfiguration(); 
 // 返回一个指定的Runner id,在不同的线程中执行时，给定阶段不改变的值。此值可用于在远程系统上创建临时资源，以确保多线程管道中同一阶段的不同实例不会彼此冲突。0 用于固有的单例。
 public int getRunnerId(); 
 // 获取为该Pipeline创建的runner（实例）的数量。
 public int getRunnerCount(); 
 // 返回SDC资源的绝对路径。 What？？？？
 public String getResourcesDirectory(); 
 // 返回Pipeline级别的常量。Immutable Map
 public Map<String, Object> getPipelineConstants();
 // 创建空记录
 // @param recordSourceId 标识记录的ID。 它应包括足够的信息以跟踪记录源。
 // 返回具有指定ID的空记录
 Record createRecord(String recordSourceId); 
 // 创建一个空记录，包括该记录的原始数据。
 // @param recordSourceId 标识记录的ID。 它应包括足够的信息以跟踪记录源。
 // @param raw the record raw data. byte[]
 // @param rawMime the MIME type of the raw data. 指定raw数据的MIME类型
 Record createRecord(String recordSourceId, byte[] raw, String rawMime);
}
```

#### 混入接口1：ConfigIssueContext

用于创建ConfigIssue实例的接口。此接口可用于所有可配置组件的上下文对象。它目前是参数化的，为Stage类提供向后兼容性。（初步判断，该接口是为配置参数操作，验证配置正确性的返回，因为在Stage和Service接口都找到了对应的证明）

```java
// 为给定的 @ConfigDef 注解描述的可配置对象，提供一个创建配置问题的功能。
public ConfigIssue createConfigIssue(String configGroup, String configName, ErrorCode errorCode, Object... args);
```

##### 证明：

在Stage和Service中，所含Context都有如下的内容

```java
public interface Context extends ProtoConfigurableEntity.Context {
 // Interface for configuration issues. Kept for backward compatibility.
 public interface ConfigIssue {}
 // 当Pipeline被初始化处理数据时，该方法被调用一次。
 // 如果stage返回一个空的stage列表。然后这个stage就可以处理数据了。
 // 否则，会认为是配置错误，或者存在问题，并且该阶段还没有准备好处理数据，因此会终止管道初始化。
 public List<ConfigIssue> init(Info info, C context);
}
```

#### 混入接口2：ELContext

定义了基于EL表达式的验证、变量设置、函数创建。

关注以下函数，定义了EL表达式可以相互引用。

```java
// 使用指定的Stage配置定义的EL函数和常量，以及指定的其他EL函数和常量，创建一个ELEval。
// @param configName stage configuration name.
// @param elDefClasses class defining additional EL functions and constants to configure the <code>ELEval</code> instance with. 定义其他EL函数和常量以配置ELEval实例的类。
public ELEval createELEval(String configName, Class<?>... elDefClasses);
```

#### 混入接口3：MetricContext

应当是和监控相关的功能内容

该接口引入第三方类库 Maven: io.dropwizard.metrics:metrics-core

```java
import com.codahale.metrics.Counter;  递增、递减的度量值
import com.codahale.metrics.Gauge;    某指标的瞬时采集值
import com.codahale.metrics.Histogram;  直方图
import com.codahale.metrics.Meter;      加权移动平均指数 该包提供1、5、15分钟的采集频率
import com.codahale.metrics.MetricRegistry;  指标注册中心，内包含注册中心提供的计算度量值
import com.codahale.metrics.Timer;

// 返回 pipeline 所使用的注册中心
    public MetricRegistry getMetrics();

// 创建一个使用管道名称和阶段实例名称加上给定名称命名的Timer。
 // @param name the <code>Timer</code> name.
 // @return A <code>Timer</code> namespaced with the pipeline name and the stage instance name plus the given name.
 public Timer createTimer(String name);

public Timer getTimer(String name);

public Meter createMeter(String name);

public Meter getMeter(String name);

public Counter createCounter(String name);

public Counter getCounter(String name);

public Histogram createHistogram(String name);

public Histogram getHistogram(String name);

public Gauge<Map<String, Object>> createGauge(String name);

public Gauge<Map<String, Object>> createGauge(String name, Comparator<String> comparator);

public Gauge<Map<String, Object>> getGauge(String name);

```

### Stage

定义公共上下文和生命周期的数据收集器Stage实现的内部基本接口。

![image-20200314150104809](./image/image-20200314150104809.png)

接口中主要定义了5个大项：

```java
public interface Stage<C extends Stage.Context> extends ProtoConfigurableEntity {
 public interface Info {} // 用于提供Stage元信息的功能
 public interface UserContext {} // 用于提供pipline或者job的开发人的相关元数据信息
 public interface Context extends ProtoConfigurableEntity.Context {} // 这里混入了ProtoConfigurableEntity的Context
 public interface ConfigIssue {}
 public List<ConfigIssue> init(Info info, C context);
 public void destory();
}

```

#### Stage.Context

在Stage当中定义的Context除混入了ProtoConfigurableEntity.Context，还扩充了哪些功能：

```java

// 返回当前环境变量的版本
    public String getEnvironmentVersion();

// 返回管道的当前执行模式。
    public ExecutionMode getExecutionMode();

/**
     * This method always returns -1 and is no longer supported.
     */
    @Deprecated
    public long getPipelineMaxMemory();

// 指示管道是否在预览模式下运行。
    public boolean isPreview();

// 返回与当前正在运行的管道关联的用户上下文。  ###看出来是有管理元数据的概念###
    public UserContext getUserContext();

// 返回Stage的Info实体
    public Info getStageInfo();

// 返回所在pipeline中的所有stages的Info信息
 public List<Info> getPipelineInfo();

// 报告错误
    public void reportError(Exception exception);

public void reportError(String errorMessage);

// 使用可本地化的错误代码和参数报告错误。
 // @param errorCode the error code to report.
 // @param args the arguments for the <code>ErrorCode</code> message template.
 public void reportError(ErrorCode errorCode, Object... args);

// 返回为Stage配置的错误处理。
    // The stage must be coded to honor this configured error handling.
    // 必须对Stage进行编码，以支持这种配置的错误处理。
    public OnRecordError getOnErrorRecord();

// Returns the time of completion of the previous batch.
    // 返回完成前一个stage的时间，如果不知道的话就为0
    public long getLastBatchTime();

// 指示在Stage处理一批记录时管道是否已停止。
    public boolean isStopped();

// 指示此Stage是否用于处理管道中的错误。
    public boolean isErrorStage();

// 创建标准的事件记录，其中包含预先填充的所需头属性。
    // @param type Type of the event (value is up to the generating stage)
    // @param version Version of the event type (to support event evolution)
    // @param recordSourceId the ID to identify the record. It should include enough information to track down the record source.
    // @return New record.
    public EventRecord createEventRecord(String type, int version, String recordSourceId);

// 创建LineageEvent并初始化常规字段。
    // @param type LineageEventType
    // @return initialized LineageEvent
    LineageEvent createLineageEvent(LineageEventType type);

// Publish given lineage event to configured lineage store.
    // @param event Lineage event that needs to be propagated to the configured lineage store.
    public void publishLineageEvent(LineageEvent event);

// Return unique id that identifies this data collector.
    // 返回标识此数据收集器的唯一id。
    public String getSdcId();

/**
     * After current batch, transition pipeline state to FINISHED.
     */
    @Deprecated
    void finishPipeline();

/**
     * After current batch, transition pipeline state to FINISHED.
     *
     * Optionally also reset the offset.
     *
     * @param resetOffset Whether offset should be discarded.
     */
    void finishPipeline(boolean resetOffset);

/**
     * Return unique id that identifies this pipeline.
     * 返回标识此管道的唯一id。
     * @return unique id that identifies this pipeline.
     */
    public String getPipelineId();

// 返回此特定Stage的所有实例跨所有Pipeline运行程序共享的并发 hash map。
 // 例如，
 // “dedup 1”阶段的所有实例将共享相同的map，而“Trash 1”阶段的所有实例也将共享相同的map，但与第一个map不同。
 // 返回map是线程安全的..由Stage实现来确保映射的内容不会被并发访问破坏。
 // 状态不是持久的，map在管道启动时总是空的。
 public Map<String, Object> getStageRunnerSharedMap(); // ??暂时不知道什么卵用，这个key应该是某个Stage的实例的名称。

// 返回为这个Stage实例特别配置的给定Service的单例实例。
 // 不同的Stage实例(甚至是相同的Stage实现)都有自己的单例实例。
 // 对该方法的多次调用将返回相同的对象。
 // @param serviceInterface Interface describing required service
 // @return Stage's own singleton instance
 public<T> T getService(Class<? extends T> serviceInterface);
```

### ToXXXContext

这些ToXXXContent有点类似

- ToSourceResponseContext

Describes methods for various context classes that deals with source response handling.

**从这个类的描述上来看，streamsets不仅仅是支持DAG，同时也支持DCG。**

```java
  void toSourceResponse(Record record);
  ```

- ToEventContext

Describes methods for various context classes that deals with error handling.

```java
  public void toEvent(EventRecord record);
  ```

- ToErrorContext

Describes methods for various context classes that deals with error handling.

```java
  public void toError(Record record, Exception exception);
  public void toError(Record record, String errorMessage);
  public void toError(Record record, ErrorCode errorCode, Object... args);
  ```

### Source

Source是数据收集器的起始Stage。原始阶段使用来自外部系统的数据，创建可以由处理器(Processor)或目标(Target)阶段处理的记录。

![image-20200312170851178](./image/image-20200312170851178.png)

### Processor

Processor是一个数据收集器处理器阶段。处理器阶段接收来自源(Source)或其他处理器阶段的记录，对这些记录执行操作并将它们写出来，以便可以由另一个处理器或目标(Target)阶段进行处理。

![image-20200312171746554](./image/image-20200312171746554.png)

### Target

Target是数据收集器的目标阶段。目标阶段接收来自源(Source)阶段或处理器(Processor)阶段的记录，并将它们写入外部系统。

![image-20200312164843344](./image/image-20200312164843344.png)

### Executor

不像Target那样持久化记录，而是在基于 外部系统 的数据中执行操作，这些数据显示在传入的记录中。

![image-20200312181116619](./image/image-20200312181116619.png)

### LineagePublisher

Data collectors collects various lineage metadata for all pipelines while they are running. One can implement subclass of LineagePublisher and configure Data Collector to use in order to propagate this data to a third party system (such as Cloudera Navigator or Oracle Metadata Manager).

数据收集器在运行时为所有Pipeline收集各种血缘元数据。可以实现LineagePublisher的子类并配置数据收集器，以便将这些数据传播到第三方系统(如Cloudera Navigator或Oracle Metadata Manager)。

```java
interface ConfigIssue {}

public interface Context {
 String getId();
	ConfigIssue createConfigIssue(ErrorCode errorCode, Object... args);
	String getConfig(String configName);
}
// 此方法在数据data collector启动期间仅调用一次。
List<ConfigIssue> init(Context context);
// 处理给定的事件列表并将它们推到血缘收集器。当且仅当所有事件都成功写入远程系统时，返回true。
boolean publishEvents(List<LineageEvent> events);

```

### Service

Base interface for Data Collector services, a reusable and configurable code that stages can depend on to outsource and share common functionality.

数据收集器服务的基本接口，一个可重用和可配置的代码，Stage可以依赖它外部Source和共享公共功能。

```java
// LineagePublisher Context that provides runtime information and services to the plugin.
// 这里的备注描述Service的Context主要为服务插件提供运行信息和服务，是要LineagePublisher功能。但却不是使用的LineagePublisher.Context，而是继承了ProtoConfigurableEntity.Context
public interface Context extends ProtoConfigurableEntity.Context {} // ???诡异的描述
// This method is called once during start up of a pipeline.
// pipeline启动时被执行调用一次。
// 如果服务返回一个ConfigIssues的空列表，则认为它已经可以使用了。
// 否则，它会被认为是配置错误，或者存在不可恢复的问题，服务还没有准备好使用，因此整个管道初始化将失败。
List<ConfigIssue> init(Context context);

void destroy();
```

### Other Service

接口层定义了一些常见的服务接口：

- SshTunnelService

Service that provides SSH Tunnel support. It initiates a forward SSH tunnel to the configured SSH server.

- LogParserService

Service for parsing logs from input streams

- DataFormatGeneratorService

Service that allows serializing records to various file formats.

- DataFormatParserService

Service for parsing various data formats from input streams.

- SdcRecordGeneratorService

Special data-format service that will always use SDC_RECORD as the output format and thus have no input parameters in the UI to configure. This is particularly helpful for error stages that always serialize SDC_RECORD.特殊的数据格式服务，它总是使用SDC_RECORD作为输出格式，因此在UI中没有需要配置的输入参数。这对于总是序列化SDC_RECORD的错误阶段特别有用。

### Interceptor

Interceptor is a plugable component similar to Stage.

与stage不同的是，它对用户不可见，而是“隐藏”在执行框架中运行。这允许拦截器执行用户不应该禁用的工作。

```java
// 主要包含3项内容
// 拦截器混入了监控上下文
public interface Context extends MetricContext {}
// 在初始化Pipeline 处理任何数据之前，将调用此方法一次。
public List<ConfigIssue> init(Context context);
// 截获记录并执行需要执行的操作。
public List<Record> intercept(List<Record> records) throws StageException;

void destroy();
```

## Exception封装

### StageException

所有Stage的异常父类

```java
private final transient ErrorCode errorCode;
private final transient ErrorMessage errorMessage;
private final transient Object[] params;
private List<AntennaDoctorMessage> antennaDoctorMessages;

...
    
public interface ErrorCode {

/**
   * Returns the error code.
   *
   * @return the error code.
   */
  public String getCode();

/**
   * Returns the built-in default message for the error code.
   *
   * @return the default message template
   */
  public String getMessage();

}

...

public class ErrorMessage implements LocalizableString {
  private static final Object[] NULL_ONE_ARG = {null};

private final String errorCode;
 private final LocalizableString localizableMessage;
 private final long timestamp;
 private final boolean preppendErrorCode;
 private final String stackTrace;
 private final List<AntennaDoctorMessage> antennaDoctorMessages;
}

...
public class AntennaDoctorMessage {
  private final String summary;
  private final String description;
}
```

![image-20200313200930257](./image/image-20200313200930257.png)

### delegate

![image-20200314121802851](./image/image-20200314121802851.png)

Client定义在ClusterJob内部，Client可以闭包访问到ClusterJob内部的元素。后续实现应该是委派模式。

### credential

A credential store gives programmatic access, via a reference name, to sensitive values such as system passwords.

凭据存储通过引用名称,提供程序对敏感值(如系统密码)的访问。

![image-20200314123329655](./image/image-20200314123329655.png)

### ext

![image-20200314125441992](./image/image-20200314125441992.png)

### EL Express

![image-20200314125619691](./image/image-20200314125619691.png)

fork/streamsets-start-asset_1

简介

发行版

贡献者

近期动态

fork/streamsets-start-asset_1 .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

fork/streamsets-start-asset_1