1.StreamSets-Concept.md · fork/streamsets-start-asset

# StreamSets 管道概念

## Pipeline Concepts and Design

- [What is a Pipeline?](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/What_isa_Pipeline.html) 什么是管道?

- [Data in Motion](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DatainMotion.html#concept_y1r_zky_bt) Data passes through the pipeline in batches. This is how it works:  运动数据批量通过管道。它是这样工作的:

- [Designing the Data Flow](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DesigningDataFlow.html#concept_ad4_gpy_bt) You can branch and merge streams in the pipeline.  设计数据流可以在管道中分支和合并流。

- [Dropping Unwanted Records](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DroppingUnwantedRecords.html#concept_ejl_lpy_bt) You can drop records from the pipeline at each stage by defining required fields or preconditions for a record to enter a stage. 通过定义记录进入某个阶段所需的字段或先决条件，可以在每个阶段将记录从管道中删除。

- [Error Record Handling](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/ErrorHandling.html#concept_pm4_txm_vq)  错误记录处理

- [Record Header Attributes](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/RecordHeaderAttributes.html#concept_wn2_jcz_dz)  记录头属性

- [Field Attributes](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/FieldAttributes.html#concept_xfm_wtp_1z) Field attributes are attributes that provide additional information about each field that you can use in pipeline logic, as needed.  字段属性是提供关于每个字段的附加信息的属性，您可以根据需要在管道逻辑中使用这些信息。

- [Processing Changed Data](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/CDC-Overview.html#concept_apw_l2c_ty) 处理更改的数据

- [Control Character Removal](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/ControlCharacters.html) 控制字符删除

- [Development Stages](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DevStages.html#concept_czx_ktn_ht)  开发Stage

- [Shortcut Keys for Pipeline Design](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/ShortcutKeys.html)  管道设计的快捷键

- [Technology Preview Functionality](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/TechPreview.html) Data Collector includes certain new features and stages with the Technology Preview designation. Technology Preview functionality is available for use in development and testing, but is not meant for use in production.

技术预览功能：数据采集器包括技术预览指定的某些新特性和阶段。技术预览功能可用于开发和测试，但不用于生产。

- [Test Origin for Preview](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/TestOrigin.html#concept_sgt_s5v_g2b) A test origin can provide test data for data preview to aid in pipeline development. In Control Hub, you can also use test origins when developing pipeline fragments. Test origins are not used when running a pipeline.

测试原点预览：测试源可以为数据预览提供测试数据，以帮助管道开发。在Control Hub中，您还可以在开发管道片段时使用测试起点。运行管道时不使用测试起点。

- [Understanding Pipeline States](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Maintenance/PipelineStates-Understanding.html)  理解 Pipeline States

[Video](https://www.youtube.com/watch?v=gvmQl5woqzc)

### Oil Pipeline

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

### Data Pipeline

![2](./image/2.png)

### Instra Pipeline

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

### Data Pipeline

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

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

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

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

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

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

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

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

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

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

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

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

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

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

# ![image-20200311210100858](./image/image-20200311210100858.png)

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

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

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

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

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

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

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

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

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

# 什么是管道？

管道描述了从原始系统到目标系统的数据流，并定义了如何沿途转换数据。

您可以使用单个[原始](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Origins_overview.html#concept_hpr_twm_jq)阶段来表示原始系统，可以使用多个[处理器](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/Processors_overview.html#concept_hpr_twm_jq)阶段来转换数据，并可以使用多个[目标](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Destinations_overview.html#concept_hpr_twm_jq) 阶段来表示目标系统。

开发管道时，可以使用[开发阶段](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DevStages.html#concept_czx_ktn_ht)来提供示例数据并生成错误以测试错误处理。您可以使用[数据预览](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Preview/DataPreview_Title.html#concept_jtn_s3m_lq)来确定阶段如何更改管道中的数据。

您可以使用[执行程序](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Executors/Executors-overview.html#concept_stt_2lk_fx)阶段执行[事件触发的任务执行](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_cph_5h4_lx)或保存事件信息。要处理大量数据，可以使用[多线程管道](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Multithreaded_Pipelines/MultithreadedPipelines.html#concept_zpp_2xc_py)或[集群模式管道](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Cluster_Mode/ClusterPipelines.html#concept_hmh_kfn_1s)。

在写入[Hive或Parquet](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)或[PostgreSQL的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/JDBC_DriftSolution/JDBC_DriftSyncSolution_title.html#concept_ljq_knr_4cb)管道中，您可以实现数据漂移解决方案，该解决方案可以检测传入数据中的漂移并更新目标系统中的表。

启动管道时，Data Collector会运行管道，直到停止管道或关闭Data Collector为止。您可以使用Data Collector运行多个管道。

在管道运行时，您可以[监视](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Monitoring/PipelineMonitoring_title.html#concept_hsp_tnt_lq)管道以验证管道是否按预期执行。您还可以定义度量标准和数据 [规则以及警报，](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Alerts/RulesAlerts_title.html#concept_ugl_nlh_q5) 以在达到特定阈值时通知您。

# 动态数据

数据分批通过管道。它是这样工作的：

原点会在从原点系统读取数据或从原点系统读取数据时创建批处理，并注意偏移量。偏移量是原点停止读取的位置。

当批次已满或超过批次等待时间限制时，源将发送批次。批次在处理器之间通过管道移动，直到到达管道目的地。

目标将批处理写入目标系统，然后Data Collector在内部提交偏移量。基于管道交付保证，Data Collector要么在将偏移写入任何目标系统后立即提交偏移，要么在收到来自所有目标系统的写入确认后立即提交偏移。偏移提交后，原始阶段将创建一个新批次。

请注意，这描述了一般的管道行为。行为可能因特定的管道配置而异。例如，对于Kafka使用者，偏移量存储在Kafka或ZooKeeper中。对于不存储数据的原始系统（例如Omniture和HTTP Client），因为不相关，所以不存储偏移量。

## 单线程和多线程管道

上面的信息描述了标准的单线程管道-原始服务器创建一个批次并将其通过管道，仅在处理前一个批次之后才创建新批次。

某些源可能会生成多个线程，以在多线程管道中启用并行处理 。在多线程管道中，您可以配置源以创建要使用的线程数或并发量。 然后，Data Collector根据管道的“ Max Runners”属性创建许多管道运行器，以执行管道处理。每个线程都连接到原始系统，创建一批数据，并将该批数据传递给可用的管道运行器。

每个管道运行程序一次处理一个批处理，就像在单个线程上运行的管道一样。当数据流减慢时，管道运行器会闲置等待，直到需要它们为止，并定期生成一个空批。您可以配置“运行者空闲时间”管道属性来指定间隔或选择退出空批次生成。

本指南中对管道的所有常规引用都描述了单线程管道，但是此信息通常适用于多线程管道。有关特定于多线程管道的更多详细信息，请参阅《[多线程管道概述》](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Multithreaded_Pipelines/MultithreadedPipelines.html#concept_zpp_2xc_py)。

## 交货保证

配置管道时，可以定义如何处理数据：是否要防止数据丢失或数据重复？

交付保证管道属性提供以下选择：

- 至少一次

确保管道处理所有数据。

如果故障导致Data Collector在处理一批数据时停止，则重新启动时将重新处理该批处理。此选项可确保没有数据丢失。

使用此选项，Data Collector 在从目标系统收到写入确认后提交偏移量。如果在Data Collector 将数据传递到目标系统之后但在收到确认并提交偏移之前发生故障，则在目标系统中最多可以复制一个批处理数据。

- 最多一次

确保不重复处理数据。

如果故障导致Data Collector在处理一批数据时停止，则在启动时，它将开始处理下一批数据。此选项可避免由于重新处理而导致目的地数据重复。

使用此选项，Data Collector 在写入后提交偏移量，而无需等待目标系统的确认。如果在Data Collector 将数据传递到目标并提交偏移量后发生故障，则可能最多不会将一批数据写入目标系统。

## 数据收集器 数据类型

在管道中， Data Collector 使用以下数据类型，在读取或写入外部系统时根据需要转换数据：

| Data Type | Description |
| :------------- | :----------------------------------------------------------- |
| BOOLEAN | True or False. Corresponds to the Boolean Java data type. |
| BYTE | 8-bit signed whole number. Ranges from -128 to 127. Corresponds to the Byte Java data type. 8位带符号整数。范围从-128到127。对应于Byte Java数据类型。 |
| BYTE_ARRAY | Array of byte values. Corresponds to the Byte[] Java data type. 字节值数组。对应于Byte [] Java数据类型。 |
| CHAR | Stores a single Unicode character. Corresponds to the Char Java data type. 存储单个Unicode字符。对应于Char Java数据类型。 |
| DATE | Time object that includes the day, month, and four-digit year. Does not include time zone. 时间对象，包括日，月和四位数的年份。不包括时区。 |
| DATETIME | Time object that includes the day, month, four-digit year, hours, minutes, and seconds. Precise to the millisecond. Does not include time zone. 包括日，月，四位数的年份，小时，分钟和秒。精确到毫秒。不包括时区。 |
| DECIMAL | Arbitrary-precision signed decimal numbers. Corresponds to the BigDecimal Java class. 任意精度带符号十进制数字。对应于BigDecimal Java类。 |
| DOUBLE | 64-bit double-precision IEEE 754 floating point. Corresponds to the Double Java data type. 64位双精度IEEE 754浮点数。对应于Double Java数据类型。 |
| FILE_REF | Internal type for [whole file format](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Formats/WholeFile.html#concept_nfc_qkh_xw). [整个文件格式的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Formats/WholeFile.html#concept_nfc_qkh_xw)内部类型。 |
| FLOAT | 32-bit single-precision IEEE 754 floating point. Corresponds to the Float Java data type. 32位单精度IEEE 754浮点。对应于Float Java数据类型。 |
| INTEGER | 32-bit signed whole number. Corresponds to the Int Java data type. 32位带符号的整数。对应于Int Java数据类型。 |
| LIST | Nested type. Contains an enumerated list of values. Values can be any type, and each value in a list can be a different type. 嵌套类型。包含值的枚举列表。值可以是任何类型，列表中的每个值可以是不同类型。 |
| LIST_MAP | Nested type. Contains a list of key-value pairs kept in a specific order. Keys are always strings. Values can be any type, and each value can be a different type. 嵌套类型。包含按特定顺序保留的键/值对列表。键始终是字符串。值可以是任何类型，每个值可以是不同的类型。 |
| LONG | 64-bit signed whole number. Ranges from -9,223,372,036,854,775,808 to 9,223,372,036,854,775,807. Corresponds to the Long Java data type. 64位签名整数。范围从-9,223,372,036,854,775,808到9,223,372,036,854,775,807。对应于Long Java数据类型。 |
| MAP | Nested type. Contains a list of key-value pairs. Keys are always strings. Values can be any type, and each value can be a different type. 嵌套类型。包含键值对列表。键始终是字符串。值可以是任何类型，每个值可以是不同的类型。 |
| SHORT | 16-bit signed whole number. Ranges from -32,768 to 32767. Corresponds to the Short Java data type. 16位带符号的整数。范围从-32,768到32767。对应于Short Java数据类型。 |
| STRING | Text. Corresponds to the String Java class. |
| TIME | Time object that includes the hours, minutes, and seconds. Precise to the millisecond. Does not include the time zone. 时间对象，包括小时，分钟和秒。精确到毫秒。不包括时区。 |
| ZONED_DATETIME | Time object that includes day, month, four-digit year, hours, minutes, seconds, and time zone. Precise to the nanosecond. Corresponds to the ZonedDateTime Java class. 时间对象，包括日，月，四位数的年份，小时，分钟，秒和时区。精确到纳秒。对应于ZonedDateTime Java类。 |

# 设计数据流

您可以在管道中分支和合并流。

## 分支流

将一个阶段连接到多个阶段时，所有数据都会传递到所有连接的阶段。您可以为阶段配置必填字段，以在记录进入阶段之前将其丢弃，但是默认情况下，所有记录都会通过。

例如，在以下管道中，来自“目录”源的所有数据都传递到管道的两个分支以进行不同类型的处理。但是您可以选择配置字段拆分器或字段替换器的必填字段，以丢弃不需要的任何记录。

![img](./image/ReplicatingStreams.png)

要基于更复杂的条件路由数据，请使用流选择器。

某些阶段会生成传递到事件流的事件。事件流源自事件生成阶段，例如起点或目的地，并从该阶段通过事件流输出传递，如下所示：

![img](./image/BranchEventStream.png)

有关事件框架和事件流的更多信息，请参见[数据流触发器概述](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_cph_5h4_lx)。

## 合并流

您可以通过将两个或多个阶段连接到同一下游阶段来合并管道中的数据流。合并数据流时，Data Collector会将所有流中的数据引导到同一阶段，但不执行流中记录的联接。

例如，在以下管道中，流选择器阶段将具有空值的数据发送到字段替换器阶段：

![img](./image/MergingStreams.png)

来自流选择器默认流的数据和来自字段替换器的所有数据将传递到表达式评估器以进行进一步处理，但顺序不特定且没有记录合并。

**重要：**管道验证不会阻止重复数据。为避免将重复数据写入目标，请配置管道逻辑以删除重复数据或防止生成重复数据。

请注意，您无法将事件流与数据流合并。事件记录必须从事件生成阶段流到目的地或执行者，而不能与数据流合并。有关事件框架和事件流的更多信息，请参见[数据流触发器概述](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_cph_5h4_lx)。

# 删除不需要的记录

您可以通过定义记录进入阶段的必填字段或前提条件，在每个阶段从管道中删除记录。

## 必填项

必填字段是必须存在于记录中的字段，以使其进入处理阶段。当记录不包括所有必填字段时，将根据为管道配置的错误处理对其进行处理。您可以为任何处理器，执行程序和大多数目标阶段定义必填字段。

将必填字段配置为整体管道逻辑的一部分，或将处理错误降至最低。例如，如果Field Hasher在/ SSN字段中编码社会保障数据。为确保所有记录都包含社会保险号，您可以将/ SSN设置为该阶段的必填字段。

## 前提条件

前提条件是记录进入处理阶段所必须满足的条件。在将记录传递到阶段或进行错误处理之前，阶段会处理所有前提条件。当一条记录不满足所有已配置的前提条件时，将根据为该阶段配置的错误处理来对其进行处理。

您可以为任何处理器，执行器和大多数目标阶段定义前提条件。您可以在前提条件中使用大多数函数，管道常量和运行时属性。

将前提条件配置为整体管道逻辑的一部分，或将处理错误降至最低。例如，您可以使用以下表达式排除来自美国以外的记录：

```
 ${record:value('/COUNTRY') == 'US'}
```

错误记录在errorMessage记录头属性中包含有关失败的前提条件的信息。当管道正在运行时，您还可以在“监视器”模式下查看最近的错误记录时查看错误消息。

# 错误记录处理

您可以在阶段级别和管道级别配置错误记录处理。您还可以指定记录的版本，以用作错误记录的基础。

当阶段在处理记录时发生错误时，Data Collector将根据阶段配置来处理记录。阶段选项之一是将记录传递到管道以进行错误处理。对于此选项，Data Collector基于管道错误记录处理配置来处理记录。

配置管道时，请注意阶段错误处理优先于管道错误处理。即，可以将管道配置为将错误记录写入文件，但是如果将阶段配置为丢弃错误记录，则将丢弃这些记录。您可能会使用此功能来减少为检查和重新处理而保存的错误记录的类型。

请注意，缺少必填字段的记录不会进入阶段。它们直接传递到管道以进行错误处理。

## 管道错误记录处理

管道错误记录处理确定了Data Collector如何处理阶段发送到管道以进行错误处理的错误记录。它还处理故意从管道中删除的记录，例如没有必填字段的记录。

管道根据“错误记录”选项卡上的“错误记录”属性处理错误记录。当Data Collector 遇到意外错误时，它将停止管道并记录该错误。

管道提供以下错误记录处理选项：

- 丢弃

管道丢弃记录。数据收集器 包括错误记录计数和指标中的记录。

- 发送回复到起源

管道将错误记录传回微服务源，以包含在对源REST API客户端的响应中。数据收集器 包括错误记录计数和指标中的记录。仅在[微服务管道中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Microservice/Microservice_Title.html#concept_qfh_xdm_p2b)使用。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入Amazon S3

管道将错误记录和相关详	细信息写入Amazon S3。数据收集器 包括错误记录计数和指标中的记录。您定义Amazon S3配置属性。![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写到另一个管道

管道将错误记录写入SDC RPC管道。数据收集器 包括错误记录计数和指标中的记录。

当您写入另一个管道时，Data Collector 有效创建一个SDC RPC原始管道，以将错误记录传递到另一个管道。

您需要创建一个SDC RPC目标管道来处理错误记录。管道必须包括SDC RPC源，该源配置为从该管道读取错误记录。

有关SDC RPC管道的更多信息，请参见[SDC RPC管道概述](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/RPC_Pipelines/SDC_RPCpipelines_title.html#concept_lnh_z3z_bt)。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入Azure事件中心

管道将错误记录和相关详细信息写入Microsoft Azure事件中心。 数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的Azure事件中心的配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写信给Elasticsearch

管道将错误记录和相关详细信息写入Elasticsearch。数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的Elasticsearch集群的配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入文件

管道将错误记录和相关详细信息写入本地目录。数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的目录和最大文件大小。错误文件是根据文件前缀管道属性命名的。

目前群集管道不支持写入文件。

- 写入Google云端存储

管道将错误记录和相关详细信息写入Google Cloud Storage。数据收集器 包括错误记录计数和指标中的记录。

您定义Google Cloud Storage配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入Google Pub / Sub

管道将错误记录和相关详细信息写入Google Pub / Sub。数据收集器 包括错误记录计数和指标中的记录。

您定义Google Pub / Sub配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写给卡夫卡

管道将错误记录和相关详细信息写入Kafka。数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的Kafka群集的配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写给Kinesis

管道将错误记录和相关详细信息写入Amazon Kinesis Streams。数据收集器 包括错误记录计数和指标中的记录。

您可以定义要使用的Kinesis流的配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入MapR流

管道将错误记录和相关详细信息写入MapR流。数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的MapR Streams集群的配置属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中无效。

- 写入MQTT

管道将错误记录和相关详细信息写入MQTT代理。数据收集器 包括错误记录计数和指标中的记录。

您定义要使用的MQTT代理的配置属性。

## 阶段错误记录处理

大多数阶段包括错误记录处理选项。处理记录时发生错误时，Data Collector将根据阶段的“常规”选项卡上的“记录错误”属性来处理记录。

阶段包括以下错误处理选项：

- 丢弃

该阶段将静默丢弃该记录。Data Collector不会记录有关错误的信息，也不会记录遇到错误的特定记录。丢弃的记录不包括在“ 监视方式”错误记录计数或指标中。

- 发送到错误

该阶段将记录发送到管道以进行错误处理。管道根据管道错误处理配置处理记录。

监视管道时，可以在该阶段的“错误记录”选项卡上查看最新的错误记录以及它们遇到的问题。停止管道后，此信息将不可用。

- 停止管道

Data Collector 停止管道并记录有关错误的信息。停止管道的错误在“ 监视器”模式下显示为警报，并在管道历史记录中显示为错误。

目前，群集模式管道不支持停止管道。

## 例

Kafka Consumer起源阶段读取的JSON数据的最大对象长度为4096个字符，并且该阶段遇到一个具有5000个字符的对象。根据阶段配置，Data Collector要么丢弃记录，停止管道，要么将记录传递给管道以进行错误记录处理。

将阶段配置为将记录发送到管道后，根据如何配置管道错误处理，将发生以下情况之一：

- 当管道丢弃错误记录时，

Data Collector

会丢弃记录，而无需注意操作或原因。

监视管道时，可以在该阶段的“错误记录”选项卡上查看最新的错误记录集以及有关错误的信息。但是，在您停止管道后，此信息将变得不可用。

- 当管道将错误记录写入目标时，Data Collector会将错误记录和其他错误信息写入目标。它还在监视器计数和指标中包括错误记录。

## 错误记录和版本

当Data Collector 创建错误记录时，它将保留触发错误的记录中的数据和属性，然后将与错误相关的信息添加为记录头属性。有关错误标题属性和与记录关联的其他内部标题属性的列表，请参见[内部属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/RecordHeaderAttributes.html#concept_itf_55z_dz)。

配置管道时，可以指定要使用的记录的版本：

- 原始记录-由原点最初生成的记录。当您想要原始记录而不进行任何其他管道处理时，请使用此记录。

- 当前记录-产生错误的阶段中的记录。根据发生的错误的类型，该记录可以在错误生成阶段进行不处理或部分处理。

当您想要保留记录导致错误之前管道完成的任何处理时，请使用此记录。

**相关概念**

[必填项](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/DroppingUnwantedRecords.html#concept_dnj_bkm_vq)

# 记录标题属性

记录标题属性是记录标题中的属性，您可以根据需要在管道逻辑中使用它们。

某些阶段会为特定目的创建记录标题属性。例如，[启用CDC的源](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/CDC-Overview.html#concept_iws_mhd_ty)在sdc.operation.type记录头属性中包括CRUD操作类型。这使启用CRUD的目标能够确定在处理记录时要使用的操作类型。类似地，Hive元数据处理器生成记录头属性，某些目的地可以将其用作[Hive漂移同步解决方案的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)一部分。

其他阶段包括在记录标题属性中与处理有关的信息以供一般使用。例如，如果您要基于该信息处理事件，则[事件生成阶段](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_zrl_mhn_lx)会将事件类型包括在记录标题属性中。几个来源包括诸如每个记录的来源文件名，位置或分区之类的信息。

您可以使用某些处理器来创建或更新记录头属性。例如，您可以使用表达式计算器来创建[基于记录的写入的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/RecordHeaderAttributes.html#concept_lmn_gdc_1w)属性。

在记录标题中包含属性不需要在管道中使用它们。例如，您可以在非CDC管道中使用启用CDC的Salesforce原点，而忽略自动生成的CDC记录头属性。

在将数据写入目标系统时，仅当使用Google Pub / Sub Publisher目标或将另一个目标使用SDC Record数据格式时，记录标题属性才会与记录一起保留。若要在使用其他数据格式时保留信息，请使用表达式计算器将信息从记录标题属性复制到记录字段。

## 使用标题属性

您可以使用表达式评估器或任何脚本处理器来创建或更新记录头属性。例如，MongoDB目标要求在记录头属性中指定CRUD操作。如果提供数据的原产地不会自动生成该信息，则可以使用表达式计算器或脚本处理器来设置属性值。

记录标题属性是字符串值。您可以在任何表达式中使用[record：attribute函数](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Expression_Language/Functions.html#concept_p1z_ggv_1r)，以在计算中包括属性值。

**要点：**记录标题属性没有字段路径。在表达式中使用属性时，请仅使用属性名称，并用引号将其括起来，因为属性是字符串，如下所示：

```
 ${record:attribute('<attribute name>')}
```

例如，下面的表达式评估器将目录原点创建的文件和偏移记录头属性添加到记录中：

![img](./image/Origins-EEval-headerattributes.png)

### 内部属性

随着记录在各个阶段之间移动，Data Collector会生成并更新一些只读内部记录头属性。可以查看这些属性以解决调试问题，但只能由Data Collector进行更新。

record：attribute函数不允许访问内部记录头属性。下表描述了内部记录头属性和可用于访问管道中数据的功能：

## 标头属性生成阶段

下表列出了生成记录标题属性以启用特殊处理的阶段：

| 阶段                                                         | 描述                                                         |
| :----------------------------------------------------------- | :----------------------------------------------------------- |
| CDC-enabled origins                                          | 在sdc.operation.type标头属性中包含CRUD操作类型，并且在记录标头属性中可以包含其他CRUD和CDC信息。有关更多信息，请参见[启用CDC的起源](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/CDC-Overview.html#concept_iws_mhd_ty)。 |
| Origins that process Avro data and the Data Parser processor | 在avroSchema记录标题属性中包括Avro模式。                     |
| Origins that process XML data                                | [启用字段XPaths](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Formats/XMLDFormat.html#concept_w3k_1ch_qz)时，[可以](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Formats/XMLDFormat.html#concept_w3k_1ch_qz)在xmlns记录头属性中包含名称空间。 |
| [Stages that generate events](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_zrl_mhn_lx) | 生成事件记录的记录标题属性。有关事件记录头属性的详细信息，请参见阶段文档中的“事件记录”。 |
| Amazon S3 origin                                             | 可以配置为[在记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/AmazonS3.html#concept_inh_qjx_yw)包括系统定义的和用户定义的 [对象元数据](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/AmazonS3.html#concept_inh_qjx_yw)。 |
| Amazon SQS Consumer origin                                   | 可以配置为在记录头属性中包括[SQS消息](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/AmazonSQS.html#concept_lk3_rfc_vbb)属性。 |
| [Azure Data Lake Storage Gen1 origin](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/ADLS-G1.html#concept_dnh_pwp_b3b) | 在记录头属性中包含有关记录的原始文件的信息。                 |
| [Azure Data Lake Storage Gen2 origin](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/ADLS-G2.html#concept_osx_qgz_xhb) | 在记录头属性中包含有关记录的原始文件的信息。                 |
| Directory origin                                             | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Directory.html#concept_tlj_3g1_2z)包含有关记录的原始文件的信息。 |
| File Tail origin                                             | 在 [记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/FileTail.html#concept_tlj_3g1_2z)包含有关记录的原始文件的信息。可以配置为对文件集使用 [标签属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/FileTail.html#concept_crd_tlx_fs)。 |
| Google Pub/Sub Subscriber origin                             | 可用时，在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PubSub.html#concept_qvl_y4q_v1b)包括用户定义的消息[属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PubSub.html#concept_qvl_y4q_v1b) |
| Groovy Scripting origin                                      | 可以配置为创建[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/GroovyScripting.html#concept_dyg_zpr_l3b)。 |
| Hadoop FS origin                                             | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/HadoopFS-origin.html#concept_efs_2fs_fdb)包含有关文件和偏移量的信息。 |
| [Hadoop FS Standalone origin](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/HDFSStandalone.html#concept_djz_pdm_hdb) | 在记录头属性中包含有关记录的原始文件的信息。                 |
| HTTP Client origin                                           | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/HTTPClient.html#concept_etl_bsh_l2b)包括响应头字段。 |
| HTTP Server origin                                           | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/HTTPServer.html#concept_rvj_5qy_4cb)包括有关请求的URL和请求标题字段的信息。 |
| JavaScript Scripting origin                                  | 可以配置为创建[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/JavaScriptScripting.html#concept_ty1_t13_p3b)。 |
| JDBC Multitable Consumer origin                              | 在[JDBC记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MultiTableJDBCConsumer.html#concept_xrx_11y_4y)包括表和数据类型信息。 |
| JDBC Query Consumer origin                                   | 可以配置为在[JDBC记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/JDBCConsumer.html#concept_egw_d4c_kw)包括表和数据类型信息 。 |
| Jython Scripting origin                                      | 可以配置为创建[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/JythonScripting.html#concept_ydl_j13_p3b)。 |
| Kafka Consumer origin                                        | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/KConsumer.html#concept_tlj_3g1_2z)包含有关记录来源的信息。 |
| Kafka Multitopic Consumer origin                             | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/KafkaMultiConsumer.html#concept_tlj_3g1_2z)包含有关记录来源的信息。 |
| MapR FS Standalone origin                                    | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MapRFS.html#concept_efs_2fs_fdb)包含有关文件和偏移量的信息。 |
| MapR Multitopic Streams Consumer origin                      | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MapRFSStandalone.html#concept_uyy_nsn_ndb)包含有关记录的原始文件的信息。 |
| MapR Streams Consumer origin                                 | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MapRStreamsMultiConsumer.html#concept_tlj_3g1_2z)包含有关记录来源的信息。 |
| MQTT Subscriber origin                                       | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MQTTSubscriber.html#concept_fwm_2kn_scb)包含有关记录来源的信息。 |
| Pulsar Consumer origin                                       | 在[记录头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PulsarConsumer.html#concept_tvr_r3m_y2b)的消息属性字段中包含信息 。 |
| RabbitMQ Consumer origin                                     | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/RabbitMQ.html#concept_rg5_yts_y1b)包含RabbitMQ [属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/RabbitMQ.html#concept_rg5_yts_y1b)。 |
| REST Service origin                                          | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/RESTService.html#concept_rrj_5qy_4cb)包括有关URL和请求头字段的信息 。 |
| Salesforce origin                                            | 在[Salesforce标头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Salesforce.html#concept_psx_1wg_cy)包括有关记录来源的Salesforce信息。 |
| SFTP/FTP/FTPS Client origin                                  | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SFTP.html#concept_tlj_3g1_2z)包含有关记录的原始文件的信息。 |
| SQL Server 2019 BDC Multitable Consumer origin               | 在[JDBC记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SQLServerBDCMultitable.html#SQLServerBDCMultitable-HeaderAtts)包括表和数据类型信息。 |
| Teradata Consumer origin                                     | 在[JDBC记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Teradata.html#concept_xrx_11y_4y)包括表和数据类型信息。 |
| Couchbase Lookup processor                                   | 在[记录标题属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/CouchbaseLookup.html#concept_am3_jjj_j3b)包含有关查找文档状态的信息。 |
| Expression Evaluator processor                               | 可以配置为创建或更新[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/Expression.html#concept_qf3_mfq_f5)。 |
| Groovy Evaluator processor                                   | 可以配置为创建或更新[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/Groovy.html#concept_uw4_kqw_1y)。 |
| Hive Metadata processor                                      | 为数据记录生成[记录头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/HiveMetadata.html#concept_g3p_sss_dw)。这些属性可作为基于[Hive](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)的[Drift Synchronization Solution的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)一部分用于基于记录的写入。可以配置为将[自定义标头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/HiveMetadata.html#concept_jv2_jjn_l1b)添加到元数据记录。 |
| HTTP Client processor                                        | 在[记录头属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/HTTPClient.html#concept_ekz_wrz_zw)包括响应头字段。 |
| JavaScript Evaluator processor                               | 可以配置为创建或更新[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/JavaScript.html#concept_sh1_khh_cy)。 |
| Jython Evaluator processor                                   | 可以配置为创建或更新[记录标题属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/Jython.html#concept_s2k_qhh_cy)。 |
| Schema Generator processor                                   | [生成模式](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/SchemaGenerator.html#concept_rfz_ks3_x1b)并将其写入用户定义的记录头属性。 |

## 基于记录的写入的记录头属性

目标可以使用记录头属性中的信息来写入数据。写入Avro数据的目标可以在记录头中使用Avro模式。Hadoop FS和MapR FS目标可以使用记录头属性来确定要写入的目录以及何时滚动文件，这是针对Hive的Drift同步解决方案的一部分。有关更多信息，请参见[Hive的漂移同步解决方案](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)。

要使用记录标题属性，请将目标配置为使用标题属性，并确保记录包含标题属性。

Hive元数据处理器自动为Hadoop FS和MapR FS生成记录头属性，以用作Hive漂移同步解决方案的一部分。对于所有其他目标，可以使用表达式评估器或脚本处理器来添加记录头属性。

您可以在目标中使用以下记录头属性：

- 所有Azure Data Lake Storage目标以及Hadoop FS，Local FS和MapR FS目标中的targetDirectory属性

targetDirectory记录标题属性定义记录写入的目录。如果目录不存在，则目的地将创建目录。targetDirectory标头属性替换目标中的“目录模板”属性。

使用targetDirectory提供目录时，为目标配置的时间基准仅用于确定记录是否延迟。时间基准不用于确定要创建的输出目录或将记录写入目录。

要使用targetDirectory标头属性，请在**输出**选项卡上的标头中 选择**目录**。

- 写入Avro数据的目标中的avroSchema属性

avroSchema标头属性定义了记录的Avro模式。使用此标头属性时，无法定义要在目标中使用的Avro模式。

要使用avroSchema标头属性，请在“ **数据格式”**选项卡上，选择**Avro**数据格式，然后为“ **Avro模式位置”** 属性选择“ **在记录头中”**。

- 所有Azure Data Lake Storage目标以及Hadoop FS，Local FS和MapR FS目标中的roll属性

当记录标题中存在roll属性时，将触发文件的滚动。

您可以定义卷标头属性的名称。当您使用Hive Metadata处理器生成滚动标题属性时，请使用默认的“滚动”属性名称。使用表达式计算器时，请使用在处理器中定义的roll属性的名称。

要使用卷标头属性，请在“ **输出”**选项卡上，选择“ **使用**卷标**属性”**并定义该**属性**的名称。

### 生成基于记录的写入的属性

您可以使用Hive元数据处理器，表达式评估器或脚本处理器为基于记录的写入生成记录头属性。Hive元数据处理器自动为Hadoop FS和MapR FS生成记录头属性，以用作Hive漂移同步解决方案的一部分。对于所有其他目标，可以使用表达式评估器或脚本处理器来添加记录头属性。

要使用Expression Evaluator或脚本处理器，您必须生成目标期望的记录头属性。使用以下准则来生成记录标题属性：

- 生成目标目录

使用表达式评估器或脚本处理器生成目标目录时，请注意以下详细信息：

目标期望该目录位于名为“ targetDirectory”的标头属性中。目标使用的目录与targetDirectory标头属性中所写的完全相同。与目录模板不同，targetDirectory属性中指定的目录不应包含任何需要评估的组件，例如运行时参数，变量或运行时属性。当您定义计算结果为目录的表达式时，可以使用任何有效组件，包括计算记录中数据的表达式。

例如，您要基于运行管道的数据收集器以及发生事务的区域和商店ID 将记录写入不同的目录。您可以设置一个名为DIR的运行时资源，该资源定义目录的基础，并为每个运行管道的Data Collector定义DIR 。然后，可以在Expression Evaluator中使用以下表达式来定义targetDirectory属性：`$ {runtime：conf（'DIR'）/ transactions / $ {record.value（'/ region'）} / $ {record.value（'/ storeID'）}`

- 生成Avro模式

使用表达式计算器或脚本处理器生成Avro模式时，请注意以下详细信息：目标期望在名为“ avroSchema”的标头属性中使用Avro模式。使用标准的Avro模式格式，例如：`{"type":"record","name":"table_name","namespace":"database_name", "fields":[{"name":"int_val","type":["null","int"],"default":null}, {"name":"str_val","type":["null","string"],"default":null}]}`数据库名称和表名称必须包含在Avro模式中。

**提示：**您可以使用Avro模式生成器来帮助生成Avro模式。

- 产生侧倾属性

使用表达式评估器或脚本处理器生成roll属性时，请注意以下详细信息：对属性使用任何名称，然后在目标位置指定属性名称。配置一个定义何时滚动文件的表达式。

要在表达式评估器中定义这些记录头属性，请执行以下步骤：

1. 在“

表达式

求值器” 的“ **表达式”**选项卡上，指定“

标题属性”

名称。

要生成目标目录，请使用`targetDirectory`。

要生成Avro模式，请使用`avroSchema`。

您可以为滚动指示器标题属性使用任何名称。

2. 对于“ **标题属性表达式”**，定义一个表达式，该表达式的计算结果为您希望目标使用的信息。

有关使用脚本处理器生成记录头属性的信息，请参阅脚本处理器文档。

## 在数据预览中查看属性

您可以使用数据预览来查看与管道中任何给定点上的记录关联的记录头属性。若要查看记录标题属性，请启用“显示记录/字段标题数据”预览属性。

例如，下图显示了由目录起源在数据预览中生成的记录。

![img](./image/HeaderAttribute-Preview.png)

“记录标题”列表在管道的这一点上显示记录中的只读内部属性集。 “值”下的标题属性是由目录来源创建的属性。

# 字段属性

字段属性是根据需要提供有关可在管道逻辑中使用的每个字段的附加信息的属性。

某些阶段会生成字段属性。例如，Salesforce原点在`salesforce.salesforceType`每个字段的属性中都包含原始Salesforce数据类型。

您可以在管道中创建，修改和评估字段属性。表达式计算器可以创建和修改字段级属性。您可以使用字段属性函数来评估字段属性。

例如，您可以使用表达式评估器基于记录数据创建字段属性，然后将记录传递到基于属性值路由数据的流选择器。

仅当在目标中使用SDC RPC数据格式时，字段属性才会自动包含在写入目标系统的记录中。

要将字段属性包括在记录数据中或在计算中使用字段属性，请使用`record:fieldAttribute` 和`record:fieldAttributeOrDefault`函数。有关字段属性函数的更多信息，请参见[记录函数](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Expression_Language/Functions.html#concept_p1z_ggv_1r)。

使用数据预览时，可以启用查看字段属性和记录标题属性以帮助管道开发。

## 使用字段属性

像记录标题属性一样，您可以使用Expression Evaluator或任何脚本处理器来创建或更新字段属性。例如，在处理Avro数据时，原点会将小数数据的精度和小数位数信息放入字段属性中。如果要在将这些值写入目标系统之前增加这些值，则可以使用Expression Evaluator或任何脚本处理器来设置属性值。

字段属性是字符串值。您还可以`record:fieldAttribute`在任何表达式中使用 函数以在计算中包括属性值。

**要点：**字段属性没有字段路径。在表达式中使用属性时，在指定关联字段的路径后，仅使用带引号的属性名称即可，如下所示：

```
 ${record:fieldAttribute(<path to the field>,'<attribute name>')}
```

例如，假设您正在作为[Avive漂移同步解决方案的](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Hive_Drift_Solution/HiveDriftSolution_title.html#concept_phk_bdf_2w)一部分来处理Avro数据。所有小数字段都包含自动生成的精度和小数位字段属性。在将数据传递到Hive元数据处理器之前，您要基于现有成本数据创建一个新的价格小数字段。

以下表达式评估器创建“价格”字段，并根据现有“成本”精度和小数位数定义精度和小数位数字段属性。

![img](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Graphics/FieldAtts-ExpressionEval.png)

## 字段属性生成阶段

下表列出了生成字段属性以启用特殊处理的阶段：

| 阶段 | 描述 |
| :----------------------------------------------------- | :----------------------------------------------------------- |
| Origins that process Avro data 处理Avro数据的来源 | 为每个十进制字段包括“精度”和“比例”字段属性。 |
| Origins that process XML data 处理XML数据的来源 | 可以在字段属性中包含字段[XPath信息](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Formats/XMLDFormat.html#concept_w3k_1ch_qz)，XML属性和名称空间声明。 |
| Google BigQuery origin | 在[field属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/BigQuery.html#concept_sjp_jy1_l3b)包括Datetime，Time和Timestamp字段的原始精度。 |
| Salesforce来源 | 在[字段属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Salesforce.html#concept_ysd_532_kz)包括数据类型信息。 |
| Salesforce查找处理器 | 在[字段属性中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/SalesforceLookup.html#concept_nl4_3q2_kz)包括数据类型信息。 |
| XML Parser | 可以在字段属性中包含字段XPath信息，XML属性和名称空间声明。 |

## 在数据预览中查看字段属性

与记录标题属性一样，您可以使用数据预览在管道

# 处理更改的数据

某些阶段使您能够轻松地在管道中处理数据更改，例如更改捕获数据（CDC）或事务数据。

启用CDC的源可以读取更改捕获数据。一些专门读取更改捕获数据，其他可以配置为读取它。当读取更改的数据时，它们确定与数据关联的CRUD操作，并将CRUD操作（例如插入，更新，向上插入或删除）包括在`sdc.operation.type`记录头属性中。

启用CRUD的处理器和目标`sdc.operation.type`在写入记录时可以在标头属性中使用CRUD操作类型，从而使外部系统能够执行适当的操作。

在管道中使用启用CDC的原始和启用CRUD的阶段，可以轻松地将更改的数据从一个系统写入另一个系统。您还可以使用启用CDC的源来写入非CRUD目的地，使用非CDC原始来写入启用CRUD的阶段。有关其工作原理的信息，请参见[用例](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Design/CDC-Overview.html#concept_y5y_5xd_5y)。

## CRUD操作标题属性

`sdc.operation.type`当读取更改的数据时，读取的具有CDC功能的原点将在所有记录中包括 记录头属性。

启用CRUD的处理器和目标`sdc.operation.type`在写入记录时可以在标头属性中使用CRUD操作类型，从而使外部系统能够执行适当的操作。

该`sdc.operation.type`记录标题属性使用下列整数表示CRUD操作：

- 1用于INSERT记录
- 2用于删除记录
- 3用于UPDATE记录
- 4个用于UPSERT记录
- 5用于不受支持的操作或代码
- UNDELETE记录为6
- 7用于替换记录
- MERGE记录的8

**注意：**根据来源系统支持的操作，某些来源仅使用操作的子集。同样，目标仅识别目标系统支持的操作子集。有关支持的操作的详细信息，请参阅来源和目标文档。

### 较早的实现

在早期版本中，使用不同的记录头属性为CDC启用了某些起源，但是现在它们都包含了 `sdc.operation.type`记录头属性。保留所有较早的CRUD标头属性以实现向后兼容。

同样，启用CRUD的目标可以在其他标头属性中查找CRUD操作类型，现在可以先查找`sdc.operation.type` 记录标头属性，然后再查找替代属性。保留备用标头属性功能是为了向后兼容。

## 启用CDC的阶段

启用CDC的阶段在`sdc.operation.type`记录头属性中提供CRUD操作类型。一些来源提供备用和附加的标题属性。

以下阶段提供CRUD记录头属性：

| 启用CDC的阶段        | CRUD记录标题属性                                             |
| :------------------- | :----------------------------------------------------------- |
| MapR DB CDC          | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。在记录标题属性中包括其他CDC信息。有关更多信息，请参见[CRUD操作和CDC标头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MapRdbCDC.html#concept_oq4_mhh_qbb)。 |
| MongoDB Oplog        | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。可以在记录标题属性（例如`op`和 `ns`属性）中包含其他CDC信息。有关更多信息，请参见[生成的记录](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MongoDBOplog.html#concept_wc3_byl_5y)。 |
| MySQL二进制日志      | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。在记录字段中包括其他CDC信息。有关更多信息，请参见[生成的记录](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MySQLBinaryLog.html#concept_rfd_15l_dy)。 |
| Oracle CDC客户端     | 在以下两个标头中均包括CRUD操作类型：`sdc.operation.type``oracle.cdc.operation`有关更多信息，请参见[CRUD操作头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/OracleCDC.html#concept_x4h_m42_5y)。在带有前缀的记录标头属性中包含[其他CDC信息](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/OracleCDC.html#concept_nn1_lxd_dx)`oracle.cdc`，例如 `oracle.cdc.table`。 |
| PostgreSQL CDC客户端 | [在记录中](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PostgreSQL.html#concept_zwv_tw5_n2b)包括CRUD操作类型[。](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PostgreSQL.html#concept_zwv_tw5_n2b)在带有前缀的记录标头属性中包含[其他CDC信息](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/PostgreSQL.html#concept_zzt_pb5_n2b)`postgres.cdc`，例如`postgres.cdc.lsn`。 |
| 销售队伍             | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。有关更多信息，请参见[CRUD操作标头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Salesforce.html#concept_yns_y2m_5y)。 |
| SQL解析器            | 在以下两个标头中均包括CRUD操作类型：`sdc.operation.type``oracle.cdc.operation`有关更多信息，请参见[生成的记录](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/SQLParser.html#concept_qwp_cwf_wdb)。 |
| SQL Server CDC客户端 | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。在名为的标头属性中包含CDC信息 `jdbc.`。有关更多信息，请参见[记录头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SQLServerCDC.html#concept_pc4_xts_r1b)。 |
| SQL Server更改跟踪   | 在`sdc.operation.type`记录头属性中包括CRUD操作类型 。在`jdbc.SYS_CHANGE`标头属性中包含来自更改表的其他信息 。有关更多信息，请参见[记录头属性](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SQLServerChange.html#concept_pc4_xts_r1b)。 |

## SQL Server或Azure SQL数据库中的数据更改

SQL Server和Azure SQL数据库提供了多种方法来跟踪数据更改。在Data Collector中，适当的来源取决于数据库用来跟踪更改的方法，如下表所示：

| 跟踪变更的方法 | 数据收集器来源                                               |
| :------------- | :----------------------------------------------------------- |
| CDC表          | [SQL Server CDC客户端](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SQLServerCDC.html#concept_ut3_ywc_v1b)有关CDC表的更多信息，请参见[Microsoft文档](https://docs.microsoft.com/en-us/sql/relational-databases/track-changes/about-change-data-capture-sql-server?view=sql-server-ver15)。 |
| 变更追踪表     | [SQL Server更改跟踪](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/SQLServerChange.html#concept_ewq_b2s_r1b)有关变更跟踪表的更多信息，请参见[Microsoft文档](https://docs.microsoft.com/en-us/sql/relational-databases/track-changes/about-change-tracking-sql-server?view=sql-server-ver15)。 |
| 时间表         | [JDBC多表使用者](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/MultiTableJDBCConsumer.html#concept_zp3_wnw_4y)或[JDBC查询使用者](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/JDBCConsumer.html#concept_qhf_hjr_bs)有关时态表的更多信息，请参见[Microsoft文档](https://docs.microsoft.com/en-us/sql/relational-databases/tables/temporal-tables?view=sql-server-ver15)。 |

## 启用CRUD的阶段

以下阶段识别存储在记录头属性中的CRUD操作，并可以基于这些值执行写操作。某些阶段还提供与CRUD相关的属性。

| 启用CRUD的阶段      | 支持的运营                                                   | 舞台处理                                                     |
| :------------------ | :----------------------------------------------------------- | :----------------------------------------------------------- |
| JDBC Tee处理器      | 插入更新删除                                                 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性包括“更改日志”属性，该属性使您能够根据您使用的启用CDC的来源处理记录。有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/JDBCTee.html#concept_qfd_tpm_5y)。 |
| Couchbase目的地     | 插入更新UPSERT删除                                           | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性该阶段中的默认写操作和不支持的操作处理属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Couchbase.html#concept_i1c_vby_g3b)。 |
| Elasticsearch目的地 | 创建（插入）更新索引（UPSERT）删除用`doc_as_upsert`（MERGE）更新 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Elasticsearch.html#concept_w2r_ktb_ry)。 |
| GPSS生产者目的地    | 插入更新合并                                                 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/GPSS.html#concept_alc_2bf_r3b)。 |
| JDBC生产者目标      | 插入更新删除                                                 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性包括“更改日志”属性，该属性使您能够根据您使用的启用CDC的来源处理记录。有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/JDBCProducer.html#concept_plv_jpn_5y)。 |
| 九都目的地          | 插入更新UPSERT删除                                           | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Kudu.html#concept_dvg_vvj_wx)。 |
| MapR DB JSON目标    | 插入更新删除                                                 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性在阶段中插入API和设置API属性有关更多信息，请参见[写入MapR DB JSON](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/MapRDBJSON.html#concept_hy5_3nb_xbb)。 |
| MongoDB目的地       | 插入更新更换删除                                             | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性在阶段中增加属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/MongoDB.html#concept_bkc_m24_4v)。 |
| Redis目的地         | 插入更新更换删除                                             | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Redis.html#concept_dz2_4xh_xbb)。 |
| Salesforce目的地    | 插入更新UPSERT删除未删除                                     | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性阶段中的默认操作和不支持的操作处理属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Salesforce.html#concept_opg_tyg_4z)。 |
| 雪花目的地          | 插入更新删除                                                 | 根据以下条件确定要使用的操作：`sdc.operation.type` 记录标题属性有关更多信息，请参见[定义CRUD操作](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Destinations/Snowflake.html#concept_umx_qgj_ggb)。 |

## 处理记录

更改日志可以提供不同格式的记录数据。JDBC Tee处理器和JDBC Producer目标可以解码大多数更改日志格式，以基于源更改日志生成记录数据。使用其他启用CRUD的目标时，可能需要向管道添加其他处理以更改记录的格式。

例如，由JDBC Query Consumer源创建的Microsoft SQL CDC记录除记录数据外，还包含记录中的CDC字段。您可以使用Field Remover处理器从记录中删除所有不必要的字段。

相反，由My SQL Binary Log源读取的MySQL Server二进制日志在New Data map字段中提供新数据或更新数据，并在Changed Data映射字段中提供更改或删除的数据。您可能想要使用Field Flattener处理器使用所需的数据来展平地图字段，并使用Field Remover处理器删除所有不必要的字段。

有关生成的记录格式的详细信息，请参阅启用CDC的原始文档。

## 用例

您可以在管道中一起或单独使用启用CDC的源和启用CRUD的目的地。以下是一些典型的用例：

- 启用CDC的源和启用CRUD的目的地

您可以使用启用CDC的源和启用CRUD的目标来轻松处理已更改的记录并将其写入目标系统。

例如，假设您要将CDC数据从Microsoft SQL Server写入Kudu。为此，您可以使用启用CDC的SQL Server CDC客户端源从Microsoft SQL Server更改捕获表中读取数据。源将CRUD操作类型放置在 `sdc.operation.type`标头属性中，在这种情况下：1表示INSERT，2表示DELETE，3表示UPDATE。您配置管道以写入启用了CRUD的Kudu目标。在Kudu目标中，可以为`sdc.operation.type` 属性中未设置任何值的任何记录指定默认操作，并且可以为无效值配置错误处理。您将默认值设置为INSERT，并将目标配置为使用此默认值表示无效值。在`sdc.operation.type` 属性中，Kudu目标为INSERT支持1，对于DELETE支持2，对于UPDATE支持3，对于UPSERT支持4。当您运行管道时，SQL Server CDC客户端起源将确定每个记录的CRUD操作类型，并将其写入 `sdc.operation.type`记录头属性。Kudu目标使用`sdc.operation.type`属性中的操作 来通知Kudu目标系统如何处理每个记录。任何在`sdc.operation.type`属性中具有未声明值的记录，例如由管道创建的记录，都将被视为INSERT记录。任何具有无效值的记录都使用相同的默认行为。

- 启用CDC的来源到非CRUD目的地

如果需要将更改的数据写入没有启用CRUD的目标的目标系统，则可以使用表达式评估器或脚本处理器将CRUD操作信息从 `sdc.operation.type`标头属性移动到字段，因此该信息将保留在记录中。例如，假设您要从Oracle LogMiner重做日志中读取并将记录与记录字段中的所有CDC信息一起写入Hive表。为此，您将使用Oracle CDC Client源读取重做日志，然后添加一个Expression Evaluator来将CRUD信息从`sdc.operation.type`标头属性中拉到记录中。Oracle CDC客户端将其他CDC信息（例如表名和scn）写入 `oracle.cdc`标头属性，因此您也可以使用表达式将该信息提取到记录中。然后，您可以使用Hadoop FS目标将增强的记录写入Hive。

- 非CDC来源到CRUD目的地

从非CDC来源读取数据时，可以使用Expression Evaluator或脚本处理器来定义 `sdc.operation.type`标头属性。

例如，假设您要从事务数据库表中读取数据，并使维度表与更改保持同步。您将使用JDBC查询使用者来读取源表，并使用JDBC查找处理器来检查维度表中每个记录的主键值。然后，根据查找处理器的输出，您知道表中是否有匹配的记录。使用表达式评估器，设置 `sdc.operation.type`记录标题属性-3以更新具有匹配记录的记录，而1以插入新记录。

当您将记录传递到JDBC Producer目标时，目标将使用`sdc.operation.type`header属性中的操作来确定如何将记录写入维度表。

# 控制字符删除

您可以使用多个阶段来从数据中删除控制字符，例如转义符或传输结束符。删除控制字符，以避免创建无效的记录。

当Data Collector删除控制字符时，它将删除ASCII字符代码0-31和127，但以下情况除外：

- 9-标签
- 10-换行
- 13-回车

您可以在以下阶段中使用“ **忽略Ctrl字符”**属性来删除控制字符：

- Amazon S3的起源
- Amazon SQS消费者来源
- Azure Data Lake Storage Gen1来源
- Azure Data Lake Storage Gen2的来源
- Azure IoT /事件中心消费者来源
- CoAP服务器来源
- 目录来源
- 文件尾源
- Google Cloud Storage的起源
- Google发布/订阅者来源
- gRPC客户端来源
- Hadoop FS起源
- Hadoop FS独立版本
- HTTP客户端来源
- HTTP服务器来源
- JMS消费者来源
- 卡夫卡消费者血统
- Kafka Multitopic消费者来源
- Kinesis消费者来源
- MapR FS来源
- MapR FS独立来源
- MapR Multitopic Streams消费者来源
- MapR流消费者来源
- MQTT订户来源
- 脉冲星消费者来源
- RabbitMQ消费者来源
- Redis消费者来源
- SFTP / FTP / FPTS客户端来源
- TCP服务器来源
- WebSocket客户端起源
- WebSocket服务器的起源
- 数据解析器处理器
- HTTP客户端处理器
- JSON解析器处理器
- 日志解析器处理器
- XML解析器处理器

# Stage 开发

您可以使用几个开发阶段来帮助开发和测试管道。

**注意：**不要在生产管道中使用开发阶段。

在开发或测试管道时，可以使用以下阶段：

- 开发数据生成器来源

生成具有指定字段名称和字段类型的记录。对于字段类型，您可以选择一种数据类型，例如String和Integer，或者一种数据类型，例如地址信息，电话号码和书名。

您可以使用“地图”或“列表地图”根字段类型。

源可以生成事件以测试事件处理功能。要生成事件记录，请选择“ **产生事件”**属性。

生成事件时，源将配置的字段用作事件记录的主体，并添加事件记录头属性。您还可以使用“ **事件名称”**属性指定事件类型。例如，要创建一个no-more-data事件，请为事件名称输入“ no-more-data”。有关事件框架和事件记录头属性的更多信息，请参见“ [数据流触发器概述”](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_cph_5h4_lx)。

源也可以生成多个线程以测试多线程管道。若要生成多个线程，请为“ **线程数”**属性输入一个大于1的 **数字**。有关多线程管道的更多信息，请参见《[多线程管道概述》](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Multithreaded_Pipelines/MultithreadedPipelines.html#concept_zpp_2xc_py)。

记录错误属性在此阶段无效。

- 开发随机记录来源

生成具有配置数量的Long字段的记录。您可以定义批处理之间的延迟以及要生成的最大记录数。

记录错误属性在此阶段无效。

- 开发原始数据源来源

根据用户提供的数据生成记录。您可以输入原始数据，选择数据的数据格式，然后配置任何与格式相关的配置选项。

例如，您可以输入一组日志数据，选择日志数据格式，然后定义该数据的日志格式和其他日志属性。

在数据预览中，此阶段显示原始源数据以及该阶段生成的数据。

源可以生成事件以测试事件处理功能。要生成事件记录，请选择“ **产生事件”**属性。

- 具有缓冲源的Dev SDC RPC

从SDC RPC目标接收记录，在将记录传递到管道的下一阶段之前，将记录临时缓冲到磁盘。用作SDC RPC目标管道中的源。**注意：** 在有意或意外停止管道之后，缓冲的记录将丢失。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中不可用。

- 开发快照重播原点

从下载的快照文件中读取记录。原点可以从快照文件中的第一组记录开始读取。或者，您可以将原点配置为从快照文件中的特定阶段开始读取。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中不可用。

- 传感器读取器原点

生成具有以下数据类型之一的记录：大气数据，例如BMxx80大气传感器生成的数据。记录包括以下字段：温度_C，压力_KPa和湿度。例如：`{"temperature_C": "12.34", "pressure_KPa": "567.89", "humidity": "534.44"}`Raspberry Pi系列单板计算机上的CPU温度数据，例如由BCM2835板载热传感器生成的数据。记录包含一个字段：temperature_C。例如：`{"temperature_C": "123.45"}`

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-SDC.png)在Data Collector管道中不可用。

- 开发者身份处理器

将所有记录传递到下一个阶段。在管道中用作占位符。您可以定义必填字段和前提条件，并配置阶段错误处理。

- 开发随机误差处理器

生成错误记录，以便您可以测试管道错误处理。您可以配置阶段以丢弃记录，定义必填字段和前提条件以及配置阶段错误处理。

- 开发记录创建器处理器

为进入阶段的每个记录生成两个记录。您可以定义必填字段和前提条件，并配置阶段错误处理。

![img](https://streamsets.com/documentation/datacollector/latest/help/reusable-content/shared-graphics/icon-Edge.png)在Data Collector Edge管道中不可用。

- 到活动目的地

生成事件以测试事件处理功能。要生成事件，请选择“ **产生事件”**属性。

目标为每个传入记录生成一个事件记录。它使用传入记录作为事件记录的主体，并添加事件记录头属性。请注意，传入记录中的任何记录头属性都可能丢失或替换。

有关事件框架和事件记录头属性的更多信息，请参见“ [数据流触发器概述”](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Event_Handling/EventFramework-Title.html#concept_cph_5h4_lx)。

# 管道设计的快捷键

设计管道时，可以使用以下快捷键：

| 快捷键              | 描述                                 |
| :------------------ | :----------------------------------- |
| Command + Z         | 在管道画布或属性面板中撤消操作。     |
| Command + Shift + Z | 在管道画布或属性面板中重做一个动作。 |

# 技术预览功能

Data Collector 包含某些带有Technology Preview名称的新功能和阶段。Technology Preview功能可用于开发和测试，但不适用于生产。

当技术预览功能被批准用于生产时，发行说明和文档会反映出更改，并且技术预览图标已从UI中删除。

技术预览阶段在预览阶段的左上角显示“技术预览”图标，如下所示：

![img](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Graphics/TechPreview-inStage.png)

当前，Data Collector 包括以下技术预览阶段：

- [Cron Scheduler原点](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/CronScheduler.html#concept_nsz_mnr_2jb) -生成具有cron表达式计划的当前日期时间的记录。
- [gRPC客户端来源](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/gRPCClient.html#concept_yp1_4zs_yfb) -通过调用gRPC服务器方法然后读取服务器返回的响应来处理gRPC服务器中的数据。源可以调用一元RPC或服务器流式RPC方法。仅在为边缘执行模式配置的管道中使用此源。
- [启动管道起点](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/StartPipeline.html#concept_h1l_xpr_2jb) -启动Data Collector，Data Collector Edge或Transformer管道。
- [Control Hub API处理器](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/ControlHubAPI.html#concept_akz_zsr_2jb) -调用Control Hub API。
- [启动作业处理器](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/StartJob.html#concept_irv_l5r_2jb) -启动Control Hub作业。
- [启动管道处理器](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Processors/StartPipeline.html#concept_bbc_cxr_2jb) -启动Data Collector，Data Collector Edge或Transformer管道。

# 测试预览的原点

测试源可以为数据预览提供测试数据，以帮助进行管道开发。在Control Hub中，开发管道片段时还可以使用测试源。运行管道时不使用测试原点。

测试源是您在管道或管道片段属性中配置的虚拟阶段。测试原点不会显示在画布中。这样，您就可以在画布中配置真实原点（要在生产中使用的原点），同时在管道或片段属性中配置测试原点。

您可以使用任何可用的原点作为测试原点。测试数据可以是测试来源允许的任何数据格式。测试数据不必与生产数据的数据格式匹配。

要在数据预览中使用测试原点，必须在数据预览配置对话框中选择测试原点作为预览源。默认情况下，数据预览使用画布中配置的原点作为预览源。

例如，假设您的IT部门为Amazon S3中的测试数据提供了单独的存储桶。在画布中，您将Amazon S3来源配置为使用实际存储桶进行生产。在管道属性中，您将虚拟Amazon S3测试来源配置为使用测试桶。

在管道开发和测试期间，您可以配置数据预览以使用测试原点，而保持生产数据不变。然后，当管道准备就绪时，您将启动管道。Data Collector 使用真实的Amazon S3原产地来处理生产数据，而忽略测试原产地。

您可以根据需要使用测试源，但是这里有一些常规方案：

- 使用测试原点访问同一原始系统中的测试数据。

在画布中使用相同类型的原点作为测试原点，并将它们配置为分别访问生产数据和测试数据以进行数据预览的测试原点。

- 使用测试源可以访问其他源系统中的测试数据。

必要时，您可以使用任何来源访问测试数据以进行数据预览。测试数据的原始位置和数据格式在传递到管道后变得无关紧要。

- 使用开发原点提供测试数据。

如果您有一些可用的测试数据，则可以使用原始数据源原点作为测试原点。或者，使用其他开发阶段来提供测试数据。

## 配置测试原点

在管道属性中配置测试原点。使用Control Hub时，还可以在管道片段属性中配置测试原点。

1. 在管道或片段属性的“ **常规”**选项卡上，选择要使用的原始类型。

您可以选择任何可用的来源类型。

2. 在“ **测试原点”**选项卡上，配置原点属性。

测试原点的原点属性与真实原点相同，所有属性都显示在单个选项卡上。

有关原点配置的详细信息，请参见“原点”一章中的“配置<origin类型>原点”。例如，有关配置目录测试源的帮助，请参阅[../Origins/Directory.html#task_gfj_ssv_yq](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/Directory.html#task_gfj_ssv_yq)。

## 在数据预览中使用测试原点

要在数据预览中使用已配置的测试原点，请配置预览配置选项。

1. 单击数据预览图标以开始数据预览。

2. 在“ **预览配置”**对话框中，将“ **预览源”**属性设置 为“ **测试原点”**，然后根据需要配置其余的数据预览属性。

有关使用数据预览的更多信息，请参见[数据预览概述](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Data_Preview/DataPreview_Title.html#concept_jtn_s3m_lq)。

# 了解管道状态

管道状态是管道的当前状态，例如“正在运行”或“已停止”。管道状态可以显示在“所有管道”列表中。管道的状态也可以出现在 Data Collector日志中。

通常在“所有管道”列表中显示以下管道状态：

- 已编辑-管道已创建或修改，并且自上次修改以来未运行。
- 已完成-管道已完成所有预期的处理，并已停止运行。
- RUN_ERROR-管道在运行和停止时遇到错误。
- 正在运行-管道正在运行。
- STOPPED-管道已手动停止。
- START_ERROR-管道在启动时遇到错误，无法启动。
- STOP_ERROR-管道在停止时遇到错误。

以下管道状态是瞬态的，很少显示在“所有管道”列表中。 当管道日志记录级别设置为Debug时，这些状态可以显示在Data Collector日志中：

- CONNECT_ERROR-运行集群模式管道时，Data Collector无法连接到基础集群管理器，例如Mesos或YARN。
- 连接-管道正在准备在Data Collector重新启动后重新启动。
- DISCONNECTED-管道与外部系统断开连接，通常是因为Data Collector正在重新启动或关闭。
- 完成-管道正在完成所有预期的处理。
- 重试-管道在运行时遇到错误后正在尝试运行。仅当将管道配置为重试错误时，才会发生这种情况。
- RUNNING_ERROR-管道在运行时遇到错误。
- 正在启动-管道正在初始化，但尚未启动。
- STARTING_ERROR-管道在启动时遇到错误。
- 停止-手动请求停止后，管道正在停止。
- STOPPING_ERROR-管道在停止时遇到错误。

[状态转换示例](https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Pipeline_Maintenance/PipelineStates-Examples.html)

fork/streamsets-start-asset

简介

发行版

贡献者

近期动态

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

简介

发行版

贡献者

近期动态

搜索帮助

fork/streamsets-start-asset