1.3. 将操作应用到连接中的数据
如果要在 Kamelet 和事件频道之间通过的数据执行一个操作,请使用 action Kamelets 作为 Kamelet Binding 中的中间步骤。例如,您可以使用 action Kamelet 序列化或反序列化数据,过滤数据,或者插入字段或消息标头。
操作操作(如过滤或添加字段)只适用于 JSON 数据(即,当 Content-Type 标头设置为 application/json时)。如果事件数据使用 JSON 以外的格式(如 Avro 或 Protocol Buffers),您必须通过添加反序列化步骤(例如,引用 protobuf- Kamelet)来转换数据的格式,例如: 该字段引用 deserialize-action protobuf-serialize-action 或 avro-serialize-action Kamelet。有关在连接中转换数据格式的更多信息,请参阅 数据转换 Kamelets。
action Kamelets 包括:
1.3.1. 在 Kamelet Binding 中添加操作
要实施一个操作 Kamelet,在 Kamelet Binding 文件的 spec 部分中,在 source 和 sink 部分之间添加一个 steps 部分。
前提条件
- 您已创建一个 Kamelet Binding,如 Kamelet Binding 中的 Connecting source 和 sink 组件 中所述。
您知道,您需要向 Kamelet Binding 和 action Kamelet 所需的参数添加哪个操作 Kamelet。
对于此流程中的示例,
predicate-filter-actionKamelet 的参数是一个字符串类型表达式,提供 JSON 路径表达式,仅过滤 coffee 的数据仅记录具有"deep"元粘贴性。请注意,predicate-filter-actionKamelet 要求您在 Kamelet Binding 中设置 Builder 特征配置属性。这个示例还包括因为事件数据格式是 JSON 导致的反序列化和序列化操作。
流程
在编辑器中打开
KameletBinding文件。例如,以下是
coffee-to-log.yaml文件的内容:apiVersion: camel.apache.org/v1alpha1 kind: KameletBinding metadata: name: coffee-to-log spec: source: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: coffee-source properties: period: 5000 sink: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: log-sink在
source部分上方添加集成部分并提供以下 Builder 特征配置属性(根据predicate-filter-actionKamelet 的要求):apiVersion: camel.apache.org/v1alpha1 kind: KameletBinding metadata: name: coffee-to-log spec: integration: traits: builder: configuration: properties: - "quarkus.arc.unremovable- types=com.fasterxml.jackson.databind.ObjectMapper" source: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: coffee-source properties: period: 5000 sink: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: log-sink
在
source和sink部分之间添加一个steps部分,并定义 action Kamelet。例如:apiVersion: camel.apache.org/v1alpha1 kind: KameletBinding metadata: name: coffee-to-log spec: integration: traits: builder: configuration: properties: - "quarkus.arc.unremovable-types=com.fasterxml.jackson.databind.ObjectMapper" source: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: coffee-source properties: period: 5000 steps: - ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: json-deserialize-action - ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: predicate-filter-action properties: expression: "@.intensifier =~ /.*deep/" - ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: json-serialize-action sink: ref: kind: Kamelet apiVersion: camel.apache.org/v1alpha1 name: log-sink- 保存您的更改。
使用
oc apply命令更新KameletBinding资源,例如:oc apply -f coffee-to-log.yamlCamel K operator re-generates 并运行它根据更新的
KameletBinding资源生成的 CamelK 集成。查看 Kamelet Binding 的状态:
oc get kameletbindings查看其对应集成的状态:
oc get integrations查看集成的日志文件输出:
kamel logs <integration-name>例如,如果集成名称为
coffee-to-log:kamel logs coffee-to-log要停止集成,请删除 Kamelet Binding:
oc delete kameletbindings/<kameletbinding-name>例如:
oc delete kameletbindings/coffee-to-log