Acceso a datos de GDS-Kafka

cdc.drs.avro

Formato interno de Huawei Cloud DRS. DRS genera datos en el formato avro usado por Kafka. GDS-Kafka puede interconectarse directamente con DRS para analizar e importar los datos.

Ninguno

drs.cdc

Para utilizar el formato avro para drs.cdc, especifique la dependencia Maven de GDS-Kafka-common y GDS-Kafka-source en los programas anteriores de Kafka y, a continuación, cree y rellene el objeto Record. Un objeto Record representa un registro de tabla. Se serializará en una matriz byte[], producida y enviada a Kafka, y utilizada por el GDS-Kafka descendente.

En el siguiente ejemplo, la tabla de destino es la tabla person del esquema public. La tabla person consta de los campos id, name y age. El op_type es de U lo que indica una operación de actualización. En este ejemplo se cambia el campo name de a a b en el registro con el ID 0 y se cambia el valor del campo age de 18 a 20.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Record record = new Record();
// Set the schema and table name of the target table.
record.setTableName("public.person");
// Set the field list.
List<Field> fields = new ArrayList<>();
fields.add(new Field("id", 0));
fields.add(new Field("name", 1));
fields.add(new Field("age", 2));
record.setFields(fields);
// Set the field value list before the table record is updated.
List<Object> before = new ArrayList<>();
before.add(new Integer(0, "0"));
before.add(new Character("utf-8", ByteBuffer.wrap("a".getBytes(StandardCharsets.UTF_8))));
before.add(new Integer(0, "18"));
record.setBeforeImages(before);
// Set the field value list after the table record is updated.
List<Object> after = new ArrayList<>();
after.add(new Integer(0, "0"));
after.add(new Character("utf-8", ByteBuffer.wrap("b".getBytes(StandardCharsets.UTF_8))));
after.add(new Integer(0, "20"));
record.setAfterImages(after);
// Set the operation type.
record.setOperation("U");
// Set the operation time.
record.setUpdateTimestamp(325943905);
// Serialize the record object into a byte[] array.
byte[] msg = Record.getEncoder().encode(record).array();

Formato avro estándar:

• El campo tableName se utiliza para describir la tabla de destino y los nombres de esquema a los que pertenece el registro actual. [Obligatorio]
• El campo operation se utiliza para describir el tipo de operación del registro actual. I indica inserción, U indica actualización y D indica eliminación. [Obligatorio]
• updateTimestamp indica la hora en que se realiza una operación en el extremo de origen. [Opcional]
• La lista beforeImages describe la información antes de actualizar o eliminar el registro actual. Los campos del before body corresponden a los de la tabla de destino. [Obligatorio para U/D]
• La lista afterImages describe la información actualizada o recién insertada del registro actual. [Obligatorio para U/D]
• La lista fields describe la lista de campos del registro de tabla actual. El índice values de los campos debe estar en la misma secuencia que los de beforeImage y afterImage. [Obligatorio]

cdc.json

En el siguiente ejemplo, la tabla de destino es la tabla person del esquema public. La tabla person consta de los campos id, name y age. El op_type es de U lo que indica una operación de actualización. En este ejemplo se cambia el campo name de a a b en el registro con el ID 1 y se cambia el valor del campo age de 18 a 20.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
"table": "public.person",
"op_type": "U",
"op_ts": "1668426344",
"current_ts": "1668426344",
"before": {
"id":"1",
"name":"a",
"age": 18
},
"after": {
"id":"1",
"name":"b",
"age": 20
}
}

Formato JSON estándar:

• El campo table describe la tabla de destino y los nombres de esquema a los que pertenece el registro actual. [Obligatorio]
• El campo op_type se utiliza para describir el tipo de operación del registro actual. I indica inserción, U indica actualización y D indica eliminación. [Obligatorio]
• op_ts indica la hora en que se realiza una operación en el extremo de origen. [Opcional]
• current_ts indica la hora a la que se importa un mensaje a Kafka. [Opcional]
• El objeto before describe la información antes de actualizar o eliminar el registro actual. Los campos del before body corresponden a los de la tabla de destino. [Obligatorio para U/D]
• La lista de objetos after describe la actualización o la información recién insertada del registro actual. [Obligatorio para U/D]

industrial.iot.json

Formato de datos de IoT:

• thing_model_name en header indica el nombre de tabla. [Obligatorio]
• Los valores de thing_id, instance_id y timestamp en header y el contenido en el cuerpo comprenden los campos del registro actual.
• Los datos de IoT son datos de series temporales y no implican actualización o eliminación. Sólo se involucran las operaciones de inserción.

industrial.iot.recursion.json

Formato de datos de IoT:

• thing_model_name en header indica el nombre de tabla. [Obligatorio]
• Los valores de thing_id, instance_id y timestamp en header y el contenido en el cuerpo comprenden los campos del registro actual.
• Los datos de IoT son datos de series temporales y no implican actualización o eliminación. Sólo se involucran las operaciones de inserción.
• En este formato de datos, la clave y el valor de body se añaden a los campos property y value en el nuevo formato para generar múltiples piezas de datos nuevos. De esta manera, las filas se convierten en columnas.

industrial.iot.event.json.independent.table

Formato de datos de flujo de eventos de IoT:

• event_name indica un nombre de tabla. [Obligatorio]
• event_id, start_time, end_time y fields comprenden el contenido de campo de un registro. [Obligatorio]
• Los datos de flujo de eventos IoT son datos de series temporales y no implican actualización ni eliminación. Sólo se involucran las operaciones de inserción.

industrial.iot.json.multi.events

Formato de datos de flujo de eventos de IoT:

• event_name indica un nombre de tabla. [Obligatorio]
• event_id, start_time, end_time y fields comprenden el contenido de campo de un registro. [Obligatorio]
• Los datos de flujo de eventos IoT son datos de series temporales y no implican actualización ni eliminación. Sólo se involucran las operaciones de inserción.

• En este formato de datos, la clave y el valor de fields se añaden a los campos field_name y field_value en el nuevo formato para generar múltiples piezas de datos nuevos. De esta manera, las filas se convierten en columnas.

insert

true (solo para tablas sin claves primarias)

No

Utilice INSERT SELECT para escribir datos de la tabla temporal a la tabla de destino.

false

Sí

Combinar datos de la tabla temporal a la tabla de destino en función de la clave principal.

No

Utilice INSERT SELECT para escribir datos de la tabla temporal a la tabla de destino.

delete

true (solo para tablas sin claves primarias)

No

Utilice INSERT SELECT para escribir datos de la tabla temporal a la tabla de destino.

false

Sí

• Si el campo delflag está definido, la combinación se realizará en función de la clave principal. Si se encuentra una clave primaria coincidente y el valor de pos en la tabla de destino es menor que el de la tabla temporal, el campo delflag se establecerá en 1. De lo contrario, se insertará un nuevo registro.
• Si el campo delflag no está definido, se encuentra una clave principal coincidente y el valor de pos en la tabla de destino es menor que el de la tabla temporal, el registro se eliminará de la tabla de destino.

No

• Si se establece el campo delflag, todos los campos de la tabla temporal se usarán para coincidir y combinar con la tabla de destino. Si se encuentra un registro coincidente y el valor de pos en la tabla de destino es menor que el de la tabla temporal, el campo delflag se establecerá en 1. De lo contrario, se insertará un nuevo registro.
• Si el campo delflag no está definido, todos los campos de la tabla temporal se usarán para coincidir con la tabla de destino. Si se encuentra un registro coincidente y el valor de pos en la tabla de destino es menor que el de la tabla temporal, el registro coincidente se eliminará de la tabla de destino.

update

true (solo para tablas sin claves primarias)

No

Utilice INSERT SELECT para escribir datos de la tabla temporal a la tabla de destino.

false

Sí

Equivalente a la operación insertar+eliminar en una tabla con una clave primaria.

No

Equivalente a la operación insertar+eliminar en una tabla sin clave primaria.