The HDFS Event Handler can  can upload data files to HDFS. These additional configuration steps are required:

The HDFS Event Handler dependencies and considerations are the same as the HDFS Handler, see HDFS Additional Considerations.

Ensure that gg.classpath includes the HDFS client libraries.

Ensure that the directory containing the HDFS core-site.xml file is in gg.classpath. This is so the core-site.xml file can be read at runtime and the connectivity information to HDFS can be resolved. For example:

gg.classpath=/{HDFSinstallDirectory}/etc/hadoop

If Kerberos authentication is enabled on the HDFS cluster, you have to configure the Kerberos principal and the location of the keytab file so that the password can be resolved at runtime:

gg.eventHandler.name.kerberosPrincipal=principal
gg.eventHandler.name.kerberosKeytabFile=pathToTheKeytabFile

You configure the HDFS Handler operation using the properties file. These properties are located in the Java Adapter properties file (not in the Replicat properties file).

To enable the selection of the HDFS Event Handler, you must first configure the handler type by specifying gg.eventhandler.name.type=hdfs and the other HDFS Event properties as follows:

The Avro Metadata Provider uses Avro schema definition files to retrieve metadata. Avro schemas are defined using JSON. For each table mapped in the process_name. prm file, you must create a corresponding Avro schema definition file.

Avro Metadata Provider Schema Definition Syntax

{"namespace": "[$catalogname.]$schemaname",
"type": "record",
"name": "$tablename",
"fields": [
     {"name": "$col1", "type": "$datatype"},
     {"name": "$col2 ",  "type": "$datatype ", "primary_key":true}, 
     {"name": "$col3", "type": "$datatype ", "primary_key":true}, 
     {"name": "$col4", "type": ["$datatype","null"]}   
   ]
}
 
namespace            - name of catalog/schema being mapped
name                 - name of the table being mapped
fields.name          - array of column names
fields.type          - datatype of the column
fields.primary_key   - indicates the column is part of primary key.

Representing nullable and not nullable columns:

"type":"$datatype" - indicates the column is not nullable, where "$datatype" is the actual datatype.
"type": ["$datatype","null"] - indicates the column is nullable, where "$datatype" is the actual datatype

The names of schema files that are accessed by the Avro Metadata Provider must be in the following format:

[$catalogname.]$schemaname.$tablename.mdp.avsc
 
$catalogname    - name of the catalog if exists
$schemaname   - name of the schema
$tablename        - name of the table
.mdp.avsc           -  constant, which should be appended always

Supported Avro Primitive Data Types

• boolean
• bytes
• double
• float
• int
• long
• string

See https://avro.apache.org/docs/1.7.5/spec.html#schema_primitive.

Supported Avro Logical Data Types

• decimal
• timestamp

Before you start the Replicat process, create Avro schema definitions for all tables mapped in Replicat's parameter file.

The Avro Metadata Provider requires no additional classpath setting.

This is an example for configuring the Avro Metadata Provider. Consider a source that includes the following table:

TABLE GG.TCUSTMER {
     CUST_CODE VARCHAR(4) PRIMARY KEY,
     NAME VARCHAR(100),
     CITY VARCHAR(200),
     STATE VARCHAR(200)
}

This table maps the(CUST_CODE (GG.TCUSTMER) in the source to CUST_CODE2 (GG_AVRO.TCUSTMER_AVRO) on the target and the column CITY(GG.TCUSTMER) in source to CITY2 (GG_AVRO.TCUSTMER_AVRO) on the target. Therefore, the mapping in the process_name. prm file is:

MAP GG.TCUSTMER, TARGET GG_AVRO.TCUSTMER_AVRO, COLMAP(USEDEFAULTS, CUST_CODE2=CUST_CODE, CITY2=CITY);
 

In this example the mapping definition is as follows:

• Source schema GG is mapped to target schema GG_AVRO.

• Source column CUST_CODE is mapped to target column CUST_CODE2.

• Source column CITY is mapped to target column CITY2.

• USEDEFAULTS specifies that rest of the columns names are same on both source and target (NAME and STATE columns).

This example uses the following Avro schema definition file:

File path: /home/ggadp/avromdpGG_AVRO.TCUSTMER_AVRO.mdp.avsc

{"namespace": "GG_AVRO",
"type": "record",
"name": "TCUSTMER_AVRO",
"fields": [
     {"name": "NAME", "type": "string"},
    {"name": "CUST_CODE2",  "type": "string", "primary_key":true},
     {"name": "CITY2", "type": "string"},
     {"name": "STATE", "type": ["string","null"]}
]
}

The configuration in the Java Adapter properties file includes the following:

gg.mdp.type = avro
gg.mdp.schemaFilesPath = /home/ggadp/avromdp

The following sample output uses a delimited text formatter with a semi-colon as the delimiter:

I;GG_AVRO.TCUSTMER_AVRO;2013-06-02 22:14:36.000000;NAME;BG SOFTWARE CO;CUST_CODE2;WILL;CITY2;SEATTLE;STATE;WA

Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) includes a sample Replicat configuration file, a sample Java Adapter properties file, and sample Avro schemas at the following location:

GoldenGate_install_directory/AdapterExamples/big-data/metadata_provider/avro

If the DDL changes in the source database tables, you may need to modify the Avro schema definitions and the mappings in the Replicat configuration file. You may also want to stop or suspend the Replicat process in the case of a metadata change event. You can stop the Replicat process by adding the following line to the Replicat configuration file (process_name. prm):

DDL INCLUDE ALL, EVENTACTIONS (ABORT)

Alternatively, you can suspend the Replicat process by adding the following line to the Replication configuration file:

DDL INCLUDE ALL, EVENTACTIONS (SUSPEND)

Avro bytes data type cannot be used as primary key.

The source-to-target mapping that is defined in the Replicat configuration file is static. Oracle GoldenGate 12.2 and later support DDL propagation and source schema evolution for Oracle Databases as replication source. If you use DDL propagation and source schema evolution, you lose the ability to seamlessly handle changes to the source metadata.

The JDBC Metadata Provider uses the JDBC driver that is provided with your target database. The DBC driver retrieves the metadata for every target table that is mapped in the Replicat properties file. Replicat processes use the retrieved target metadata to map columns.

You can enable this feature for JDBC Handler by configuring the REPERROR property in your Replicat parameter file. In addition, you need to define the error codes specific to your RDBMS JDBC target in the JDBC Handler properties file as follows:

gg.error.duplicateErrorCodes

gg.error.notFoundErrorCodes

gg.error.deadlockErrorCodes 

For example:

#ErrorCode
gg.error.duplicateErrorCodes=1062,1088,1092,1291,1330,1331,1332,1333
gg.error.notFoundErrorCodes=0
gg.error.deadlockErrorCodes=1213

To understand how the various JDBC types are mapped to database-specific SQL types, see https://docs.oracle.com/javase/6/docs/technotes/guides/jdbc/getstart/mapping.html#table1.

The following are the configurable values for the JDBC Metadata Provider. These properties are located in the Java Adapter properties file (not in the Replicat properties file).

MySQL Driver Configuration

gg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:oracle:thin:@myhost:1521:orcl
gg.mdp.DriverClassName=oracle.jdbc.driver.OracleDriver
gg.mdp.UserName=username
gg.mdp.Password=password

Netezza Driver Configuration

gg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:netezza://hostname:port/databaseName
gg.mdp.DriverClassName=org.netezza.Driver
gg.mdp.UserName=username
gg.mdp.Password=password

Oracle OCI Driver configuration

ggg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:oracle:oci:@myhost:1521:orcl
gg.mdp.DriverClassName=oracle.jdbc.driver.OracleDriver
gg.mdp.UserName=username
gg.mdp.Password=password

Oracle Teradata Driver configuration

gg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:teradata://10.111.11.111/USER=username,PASSWORD=password
gg.mdp.DriverClassName=com.teradata.jdbc.TeraDriver
gg.mdp.UserName=username
gg.mdp.Password=password

Oracle Thin Driver Configuration

gg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:mysql://localhost/databaseName?user=username&password=password
gg.mdp.DriverClassName=com.mysql.jdbc.Driver
gg.mdp.UserName=username
gg.mdp.Password=password

Redshift Driver Configuration

gg.mdp.type=jdbc
gg.mdp.ConnectionUrl=jdbc:redshift://hostname:port/databaseName
gg.mdp.DriverClassName=com.amazon.redshift.jdbc42.Driver
gg.mdp.UserName=username
gg.mdp.Password=password

The Hive Metadata Provider uses both Hive JDBC and HCatalog interfaces to retrieve metadata from the Hive metastore. For each table mapped in the process_name.prm file, a corresponding table is created in Hive.

The default Hive configuration starts an embedded, local metastore Derby database. Because, Apache Derby is designed to be an embedded database, it allows only a single connection. The limitation of the Derby Database means that it cannot function when working with the Hive Metadata Provider. To workaround this limitation this, you must configure Hive with a remote metastore database. For more information about how to configure Hive with a remote metastore database, see https://cwiki.apache.org/confluence/display/Hive/AdminManual+Metastore+Administration.

Hive does not support Primary Key semantics, so the metadata retrieved from Hive metastore does not include a primary key definition. When you use the Hive Metadata Provider, use the Replicat KEYCOLS parameter to define primary keys.

KEYCOLS

Use the KEYCOLS parameter must be used to define primary keys in the target schema. The Oracle GoldenGate HBase Handler requires primary keys. Therefore, you must set primary keys in the target schema when you use Replicat mapping with HBase as the target.

The output of the Avro formatters includes an Array field to hold the primary column names. If you use Replicat mapping with the Avro formatters, consider using KEYCOLS to identify the primary key columns.

For example configurations of KEYCOLS, see Review a Sample Configuration.

Supported Hive Data types

• BIGINT

• BINARY

• BOOLEAN

• CHAR

• DATE

• DECIMAL

• DOUBLE

• FLOAT

• INT

• SMALLINT

• STRING

• TIMESTAMP

• TINYINT

• VARCHAR

See https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Types.

You can find a list of supported databases that you can use to configure remote Hive metastore can be found at https://cwiki.apache.org/confluence/display/Hive/AdminManual+MetastoreAdmin#AdminManualMetastoreAdmin-SupportedBackendDatabasesforMetastore.

The following example shows a MySQL database is configured as the Hive metastore using properties in the ${HIVE_HOME}/conf/hive-site.xml Hive configuration file.

<property>
         <name>javax.jdo.option.ConnectionURL</name>	
         <value>jdbc:mysql://MYSQL_DB_IP:MYSQL_DB_PORT/DB_NAME?createDatabaseIfNotExist=false</value>
 </property>
 
 <property>
        <name>javax.jdo.option.ConnectionDriverName</name>
        <value>com.mysql.jdbc.Driver</value>
 </property>
 
 <property>
          <name>javax.jdo.option.ConnectionUserName</name>
     <value>MYSQL_CONNECTION_USERNAME</value>
 </property>
 
 <property>
         <name>javax.jdo.option.ConnectionPassword</name>
         <value>MYSQL_CONNECTION_PASSWORD</value>
 </property>

To see a list of parameters to configure in the hive-site.xml file for a remote metastore, see https://cwiki.apache.org/confluence/display/Hive/AdminManual+MetastoreAdmin#AdminManualMetastoreAdmin-RemoteMetastoreDatabase.

For the Hive Metadata Provider to connect to Hive, you must configure thehive-site.xml file and the Hive and HDFS client jars in the gg.classpath variable. The client JARs must match the version of Hive to which the Hive Metadata Provider is connecting.

For example, if the hive-site.xml file is created in the /home/user/oggadp/dirprm directory, then gg.classpath entry is gg.classpath=/home/user/oggadp/dirprm/

1. Create a hive-site.xml file that has the following properties:

   <configuration>
   <!-- Mandatory Property --> 
   <property>
   <name>hive.metastore.uris</name>
   <value>thrift://HIVE_SERVER_HOST_IP:9083</value>
   <property>
    
   <!-- Optional Property. Default value is 5 -->
   <property>
   <name>hive.metastore.connect.retries</name>
   <value>3</value>
   </property>
    
   <!-- Optional Property. Default value is 1 -->
   <property>
   <name>hive.metastore.client.connect.retry.delay</name>
   <value>10</value>
   </property>
    
   <!-- Optional Property. Default value is 600 seconds -->
   <property>
   <name>hive.metastore.client.socket.timeout</name>
   <value>50</value>
   </property>
   
    </configuration>

2. By default, the following directories contain the Hive and HDFS client jars:

   HIVE_HOME/hcatalog/share/hcatalog/*
   HIVE_HOME/lib/*
   HIVE_HOME/hcatalog/share/webhcat/java-client/*
   HADOOP_HOME/share/hadoop/common/*
   HADOOP_HOME/share/hadoop/common/lib/*
   HADOOP_HOME/share/hadoop/mapreduce/*
   

   Configure the gg.classpath exactly as shown in the step 1. The path to the hive-site.xml file must be the path with no wildcard appended. If you include the * wildcard in the path to the hive-site.xml file, it will not be located. The path to the dependency JARs must include the * wildcard character to include all of the JAR files in that directory in the associated classpath. Do not use *.jar.

This is an example for configuring the Hive Metadata Provider. Consider a source with following table:

TABLE GG.TCUSTMER {
     CUST_CODE VARCHAR(4)   PRIMARY KEY,
     NAME VARCHAR(100),
     CITY VARCHAR(200),
     STATE VARCHAR(200)}

The example maps the column CUST_CODE (GG.TCUSTMER) in the source to CUST_CODE2 (GG_HIVE.TCUSTMER_HIVE) on the target and column CITY (GG.TCUSTMER) in the source to CITY2 (GG_HIVE.TCUSTMER_HIVE)on the target.

Mapping configuration in the process_name. prm file includes the following configuration:

MAP GG.TCUSTMER, TARGET GG_HIVE.TCUSTMER_HIVE, COLMAP(USEDEFAULTS, CUST_CODE2=CUST_CODE, CITY2=CITY) KEYCOLS(CUST_CODE2); 

In this example:

• The source schema GG is mapped to the target schema GG_HIVE.

• The source column CUST_CODE is mapped to the target column CUST_CODE2.

• The source column CITY is mapped to the target column CITY2.

• USEDEFAULTS specifies that rest of the column names are same on both source and target (NAME and STATE columns).

• KEYCOLS is used to specify that CUST_CODE2 should be treated as primary key.

Because primary keys cannot be specified in the Hive DDL, the KEYCOLS parameter is used to specify the primary keys.

You can create the schema and tables for this example in Hive by using the following commands. You can create the schema and tables for this example in Hive by using the following commands. To start the Hive CLI use the following command:

HIVE_HOME/bin/hive

To create the GG_HIVE schema, in Hive, use the following command:

hive> create schema gg_hive;
OK
Time taken: 0.02 seconds

To create the TCUSTMER_HIVE table in the GG_HIVE database, use the following command:

hive> CREATE EXTERNAL TABLE `TCUSTMER_HIVE`(
    >   "CUST_CODE2" VARCHAR(4),
    >   "NAME" VARCHAR(30),
    >   "CITY2" VARCHAR(20),
    >   "STATE" STRING);
OK
Time taken: 0.056 seconds

Configure the .properties file in a way that resembles the following:

gg.mdp.type=hive
gg.mdp.connectionUrl=jdbc:hive2://HIVE_SERVER_IP:10000/gg_hive
gg.mdp.driverClassName=org.apache.hive.jdbc.HiveDriver

The following sample output uses the delimited text formatter, with a comma as the delimiter:

I;GG_HIVE.TCUSTMER_HIVE;2015-10-07T04:50:47.519000;cust_code2;WILL;name;BG SOFTWARE CO;city2;SEATTLE;state;WA

A sample Replicat configuration file, Java Adapter properties file, and Hive create table SQL script are included with the installation at the following location:

GoldenGate_install_directory/AdapterExamples/big-data/metadata_provider/hive

You can secure the Hive server using Kerberos authentication. For information about how to secure the Hive server, see the Hive documentation for the specific Hive release. The Hive Metadata Provider can connect to a Kerberos secured Hive server.

Make sure that the paths to the HDFS core-site.xml file and the hive-site.xml file are in the handler's classpath.

Enable the following properties in the core-site.xml file:

<property>
<name>hadoop.security.authentication</name>
<value>kerberos</value> 
</property>
 
<property> 
<name>hadoop.security.authorization</name> 
<value>true</value> 
</property>

Enable the following properties in the hive-site.xml file:

<property>
<name>hive.metastore.sasl.enabled</name>
<value>true</value>
</property>
 
<property>
<name>hive.metastore.kerberos.keytab.file</name>
<value>/path/to/keytab</value> <!-- Change this value -->
</property>
 
<property>
<name>hive.metastore.kerberos.principal</name>
<value>Kerberos Principal</value> <!-- Change this value -->
</property>
 
<property>
   <name>hive.server2.authentication</name>
    <value>KERBEROS</value>
</property>
 
<property>
   <name>hive.server2.authentication.kerberos.principal</name>
    <value>Kerberos Principal</value> <!-- Change this value -->
</property>
 
<property>
    <name>hive.server2.authentication.kerberos.keytab</name>
    <value>/path/to/keytab</value> <!-- Change this value -->
</property>

Tables in Hive metastore should be updated, altered, or created manually if the source database tables change. In the case of a metadata change event, you may wish to terminate or suspend the Replicat process. You can terminate the Replicat process by adding the following to the Replicat configuration file (process_name. prm):

DDL INCLUDE ALL, EVENTACTIONS (ABORT)

You can suspend, the Replicat process by adding the following to the Replication configuration file:

DDL INCLUDE ALL, EVENTACTIONS (SUSPEND)

Columns with binary data type cannot be used as primary keys.

The source-to-target mapping that is defined in the Replicat configuration file is static. Oracle GoldenGate 12.2 and later versions supports DDL propagation and source schema evolution for Oracle databases as replication sources. If you use DDL propagation and source schema evolution, you lose the ability to seamlessly handle changes to the source metadata.

The most common problems encountered are the Java classpath issues. The Hive Metadata Provider requires certain Hive and HDFS client libraries to be resolved in its classpath.

The required client JAR directories are listed in Classpath Configuration. Hive and HDFS client JARs do not ship with Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA). The client JARs should be of the same version as the Hive version to which the Hive Metadata Provider is connecting.

To establish a connection to the Hive server, the hive-site.xml file must be in the classpath.

If the mapped target table is not present in Hive, the Replicat process will terminate with a "Table metadata resolution exception".

For example, consider the following mapping:

MAP GG.TCUSTMER, TARGET GG_HIVE.TCUSTMER_HIVE, COLMAP(USEDEFAULTS, CUST_CODE2=CUST_CODE, CITY2=CITY) KEYCOLS(CUST_CODE2);

This mapping requires a table called TCUSTMER_HIVE to be created in the schema GG_HIVE in the Hive metastore. If this table is not present in Hive, then the following exception occurs:

ERROR [main) - Table Metadata Resolution Exception
Unable to retrieve table matadata. Table : GG_HIVE.TCUSTMER_HIVE
NoSuchObjectException(message:GG_HIVE.TCUSTMER_HIVE table not found)

The following table lists the Google BigQuery datatypes that are supported and their default scale and precision values:

The following table lists the Google BigQuery datatypes that are supported and their default scale and precision values:

The BigQuery datatypes that are not supported by metadata provider are complex datatypes, such as GEOGRAPHY, JSON, ARRAY, INTERVAL, and STRUCT. The metadata provider is going to abend with invalid datatype exception if it encounters them.

The following table lists the configuration properties for BigQuery metadata provider:

The Avro Row Formatter formats operation data from the source trail file into messages in an Avro binary array format. Each individual insert, update, delete, and truncate operation is formatted into an individual Avro message. The source trail file contains the before and after images of the operation data. The Avro Row Formatter takes the before-image and after-image data and formats it into an Avro binary representation of the operation data.

The Avro Row Formatter formats operations from the source trail file into a format that represents the row data. This format is more compact than the output from the Avro Operation Formatter for the Avro messages model the change data operation.

The Avro Row Formatter may be a good choice when streaming Avro data to HDFS. Hive supports data files in HDFS in an Avro format.

This section contains the following topics:

The Avro Operation Formatter formats operation data from the source trail file into messages in an Avro binary array format. Each individual insert, update, delete, and truncate operation is formatted into an individual Avro message. The source trail file contains the before and after images of the operation data. The Avro Operation Formatter formats this data into an Avro binary representation of the operation data.

This format is more verbose than the output of the Avro Row Formatter for which the Avro messages model the row data.

CloudEvents is a specification for describing event data in common formats to provide interoperability across services, platforms and systems. CloudEvents format defines the list of attributes to describe the event, essentially an envelope with a set of mandatory and optional attributes.

Currently, CloudEvents format is limited to the JSON event format.

When CloudEvents format is enabled in the Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) targets, the final JSON records will look like as follows where data field contains the original data records.

For Insert:

For Updates:

For Deletes:

{"specversion":"1.0","id":"OGG-00000000000000005272","source":"uri:slcaa318::oldhome:qastaf:ggs_home:yuga:ora:v122_rc3:ETEST","type":"QASOURCE.TCUSTORD","datacontenttype":"application/json","data":{"table":"QASOURCE.TCUSTORD","op_type":"D","op_ts":"2015-11-05 18:45:39.000000","current_ts":"2024-07-30 05:52:26.052000","pos":"00000000000000005272","before":{"CUST_CODE":"DAVE","ORDER_DATE":"1993-11-03 07:51:35","PRODUCT_CODE":"PLANE","ORDER_ID":600,"PRODUCT_PRICE":135000.00,"PRODUCT_AMOUNT":2,"TRANSACTION_ID":200}}}

CloudEvents formatter mandates the following four attributes for each event record.

"id", "source", "specversion", "type"

specversion

This is the non-empty Cloud Events specification version string, and it is “1.0” in the current release.

For example: "specversion":"1.0"

id

This is the unique identifier for the event records. It is defined as a non-empty string. By default, the position of the record prefixed with “OGG-” is used for event record ID.

For example: "id":"OGG-00000000000000005100"

This can be overridden by using the property idmappingtemplate.

For example: gg.handler.kafkahandler.format.idmappingtemplate=OGG-${position}

type

This is the type of the event related to the originating occurrence. Defaulting to com.oracle.goldengate for GG for DAA and can be overridden by using the property typemappingtemplate.

For example: gg.handler.kafkahandler.format.typemappingtemplate=${tablename}

source

The source field is the context in which the occurrence happened, and it is in the format of URI-Reference. The source trail producer obtained from the source trail header file is used as this field value.

For example: "source":"uri:slcaa318::oldhome:qastaf:ggs_home:yuga:ora:v122_rc3:ETEST"

This can be overridden by using the property sourcemappingtemplate.

data

The data field contains the original event data. It can only be in JSON object type. Defaulted to represent the data it in Operation Data format.

Additional attributes:

Additional attributes can be sent upon the configuration of the specified explicit properties.

Following are the additional attributes.

Datacontenttype:

The valid value for now is “application/json”.

Subject:

Subject of the event in the context of the event producer. Can be configured using subjectMappingTemplate property.

For example: gg.handler.kafkahandler.format.subjectMappingTemplate=OGGDAA-KafkaStreams

Time

Timestamp of when the occurrence happened. Can be controlled by template keywords.

For example: TimeMappingTemplategg.handler.kafkahandler.format.timeMappingTemplate =${opTimestamp}

The Cloud Event formatter represents the event data in Operation Data Formatting where the data is represented by before and after members that are objects. These objects contain members whose keys are the column names and whose values are the column values.

Operation data is modeled as follows:

• Inserts: Includes the after-image data.
• Updates: Includes both the before-image and the after-image data.
• Deletes: Includes the before-image data.

For example:

Row data is modeled as follows:

• Inserts: Includes the after-image data.
• Updates: Includes the after-image data.
• Deletes: Includes the before-image data.

An Avro schema for a designated table is published the first time an operation pertaining to that table is detected. Within the generated schema, the title and id attributes play a crucial role in identifying its connection to the Cloud Event formatter. The title attribute denotes the specific table, whereas the id attribute features the fully qualified table name, prefixed with ogg:dataStream:DMLRecord:.

{
    "$schema": "http://json-schema.org/draft-04/schema#",
    "title": "Oracle GoldenGate JSON Schema for DML Record from QASOURCE.TCUSTMER",
    "description": "JSON schema for table QASOURCE.TCUSTMER",
    "id": "ogg:dataStream:DMLRecord:QASOURCE.TCUSTMER",
    "definitions": {
        "row": {
            "type": "object",
            "properties": {
                "CUST_CODE": {
                    "type": [
                        "string",
                        "null"
                    ]
                },
                "NAME": {
                    "type": [
                        "string",
                        "null"
                    ]
                },
                "CITY": {
                    "type": [
                        "string",
                        "null"
                    ]
                },
                "STATE": {
                    "type": [
                        "string",
                        "null"
                    ]
                }
            },
            "additionalProperties": false
        },
        "tokens": {
            "type": "object",
            "description": "Token keys and values are free form key value pairs.",
            "properties": {
            },
            "additionalProperties": true
        }
    },
    "type": "object",
    "properties": {
        "table": {
            "description": "The MetaColumn table",
            "type": "string"
        },
        "op_type": {
            "description": "The MetaColumn op_type",
            "type": "string"
        },
        "op_ts": {
            "description": "The MetaColumn op_ts",
            "type": "string"
        },
        "current_ts": {
            "description": "The MetaColumn current_ts",
            "type": "string"
        },
        "pos": {
            "description": "The MetaColumn pos",
            "type": "string"
        },
        "before": {
            "$ref": "#/definitions/row"
        },
        "after": {
            "$ref": "#/definitions/row"
        }
    },
    "required": [
        "table",
        "op_type",
        "op_ts",
        "current_ts",
        "pos"
    ],
    "additionalProperties": false
}

gg.mdp.type=avro
gg.mdp.schemaFilesPath=

The Existing Avro Schema formatter can generate messages in following formats:

• Raw avro – Generally used with Kafka or other integrations when sending individual messages.
• Avro OCF (object container file) – Generally used when generating files such as with the File Writer Handler. This is the format which has the Avro schema in the file header and then contains one or more records which conform to the schema.
• JSON – It is possible to convert the Avro Generic Record to JSON and publish the JSON. This can be used for debugging since JSON provides human readability whereas Avro does not. Additionally, it could be used if you wish to send data in JSON formats which support nested hierarchical structures.

The Existing Avro Schema Formatter supports mapping into Avro hierarchical structures. This is one of the powerful features of the Existing Avro Schema formatter. As previously stated, the GoldenGate trail structure is a flat structure of key-value pairs which model the flat column name and column value structure of an RDBMS. Avro can be a hierarchical structure where records can be nested inside of fields and records. The Avro Metadata provider has been modified to traverse the Avro schema and to flatten the Avro structure to return the target metadata to the GoldenGate mapping.

{
  "type" : "record",
  "name" : "TCUSTMERNEST",
  "namespace" : "QASOURCE",
  "fields" : [ {
    "name" : "CUST_CODE",
    "type" : [ "null", "string" ],
    "default" : null,
    "primary_key": true
  }, {
    "name" : "SUB",
    "type" : {
      "type" : "record",
      "name" : "subrecord",
      "fields" : [ {
        "name" : "NAME",
        "type" : [ "null", "string" ],
        "default" : null
      }, {
        "name" : "STATE",
        "type" : [ "null", "string" ],
        "default" : null
      }, {
        "name" : "CITY",
        "type" : [ "null", "string" ],
        "default" : null
      } ]
    }
  } ]
}

CUST_CODE
SUB-NAME
SUB-STATE
SUB-CITY

MAP QASOURCE.TCUSTMER, TARGET QASOURCE.TCUSTMERNEST COLMAP (CUST_CODE=CUST_CODE, SUB-NAME=NAME, SUB-CITY=CITY, SUB-STATE=STATE);

Avro has no restriction how deep fields can be nested. It can be confusing to customers who are looking at an Avro schema to understand what the flattened structure looks like. To help the flattened structure is logged to the <replicat name>.log file. Therefore, if you run the replicat, then you can examine the log file to see the flattened fields.

INFO  2023-08-09 14:54:42.000242 [main] -
Retrieved Avro metadata for table [QASOURCE.TCUSTMERNEST]
    Field name [CUST_CODE] Type [STRING] Nullable [true].
    Field name [SUB-NAME] Type [STRING] Nullable [true].
    Field name [SUB-STATE] Type [STRING] Nullable [true].
    Field name [SUB-CITY] Type [STRING] Nullable [true].
End retrieved metadata.

• Column has a value.
• Column value is null.
• Column value is missing.

• If the mapped Avro field is nullable, then null will be mapped to the target field.
• If the mapped Avro field is not nullable, then the default value for the field will be mapped as defined in the Avro schema.
• If the target field is not nullable and provides no default value, then this is an exception. The replicat will log an exception and abend.

• If a default value is defined in the Avro schema then the default value will be mapped into the target field.
• If no default value is defined in the Avro schema, and the field is nullable, then null will be mapped into the target field.
• If the target field is not nullable and provides no default value, then this is an exception. The replicat will log an exception and abend.

Additionally, it is important to understand that the default value for a field may be null.

Examples of Avro field definitions:

{"name": "STRING_REQUIRED", "type": "string"}

{"name": "STRING_REQUIRED", "type": "string", "default": "mydefault"}

{"name": "STRING_NULLABLE", "type": ["null","string"]}

{"name": "STRING_NULLABLE", "type": ["null","string"], "default": null}

String field nullable with non-null default.

{"name": "STRING_NULLABLE", "type": ["string","null"], "default": "mydefault"}

Note, it is an Avro requirement that default value apply to the first type in the type union. None of the of the following are legal. These will fail when the Avro library parses the schema and replicat will abend.

{"name": "STRING_NULLABLE", "type": ["string","null"], "default": null}
{"name": "STRING_NULLABLE", "type": ["null","string"], "default": "mydefault"}
{"name": "STRING_REQUIRED", "type": "string", "default": null}

Maps and Arrays

Neither the Avro Metadata Provider nor the Existing Avro schema formatter support Avro arrays or maps. Avro arrays and maps are containers, which can hold 0 to N elements. The Oracle GoldenGate trail structure is a flat structure of key and value pairs which in turn models the flat column name and column value structure of an RDBMS. Target metadata must be made to be a flat structure and the number of elements must be deterministic. Arrays and maps are problematic because they can contain 0 to N elements. Different messages can contain different numbers of elements for the collections.

Recursive Record Definitions

Avro supports recursive record definitions. A recursive record definition is where an element in record A contains a field which references record B. The definition of record B contains a field which references record A. The following is an example of a recursive record definition.

 {
  "type" : "record",
  "name" : "TCUSTMERINVALID",
  "namespace" : "QASOURCE",
  "fields" : [ {
    "name" : "CUST_CODE",
    "type" : [ "null", "string" ],
    "default" : null,
    "primary_key": true
  }, {
    "name" : "SUB1",
    "type" : [ "null", {
      "type" : "record",
      "name" : "subrecord",
      "fields" : [ {
        "name" : "NAME",
        "type" : [ "null", "string" ],
        "default" : null
      }, {
        "name" : "SUB2",
        "type" : {
          "type" : "record",
          "name" : "subrecord2",
          "fields" : [ {
            "name" : "RECUSIVEELEMENT",
            "type" : [ "null", "subrecord" ],
            "default" : null
          } ]
        }
      } ]
    } ],
    "default" : null
  }, {
    "name" : "STATE",
    "type" : [ "null", "string" ],
    "default" : null
  } ]
}

ERROR 2023-08-09 15:16:19.000560 [main] - The Avro schema for table [QASOURCE.TCUSTMERINVALID] encountered a recursive reference to a record schema.  The recursively referenced record schema is namespace [QASOURCE] name [subrecord2].  Schemas with recursively referenced record schemas are not supported.
ERROR 2023-08-09 15:16:19.000563 [main] - AVRO-00005 Unable to retrieve table matadata table : [QASOURCE.TCUSTMERINVALID].
oracle.goldengate.datasource.metadata.provider.ResolutionException: The Avro schema for table [QASOURCE.TCUSTMERINVALID] encountered a recursive reference to a record schema.  The recursively referenced record schema is namespace [QASOURCE] name [subrecord2].  Schemas with recursively referenced record schemas are not supported.

Non-discrete Unions

The Avro MDP only supports Unions of null and another type. For example: "type" : [ "null", "string" ], The following is an Avro union type of double and string. This is legal is Avro; however, it cannot be supported in the Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) metadata provider. The problem is that the field needs to be assigned to a discrete type. This field type, by definition, is not discrete. It can be either a double or a string: "type" : [ "double", "string" ],

Metacolumns Configuration

Metacolumn configuration is not supported with the existing Avro schema formatter. The following configuration will cause the replicat process to abend: gg.handler.name.format.metaColumnsTemplate=

The reason is that the input data from replicat is being mapped into an existing Avro schema. The result of configuring metacolumns is to add additional fields to the output data which is contrary to the idea of outputting the data which conforms to the provided Avro schema. It is possible to output operation metadata into the output records. To accomplish this the user should modify the Avro schema to add one or more fields to hold the desired operation metadata. Then the replicat mapping statement can be used to map operation metadata into the fields.

Sample Configuration

The following is example configuration of end to end configuration using the existing Avro schema formatter functionality.

REPLICAT mdp
-- Trail file for this example is located in "AdapterExamples/trail" directory
-- Command to add REPLICAT
-- add replicat mdp, exttrail AdapterExamples/trail/tr
TARGETDB LIBFILE libggjava.so SET property=dirprm/mdp.props
REPORTCOUNT EVERY 1 MINUTES, RATE
GROUPTRANSOPS 1000
MAP QASOURCE.TCUSTMER, TARGET QASOURCE.TCUSTMERNEST COLMAP (SUB-NAME=NAME, SUB-CITY=CITY, SUB-STATE=STATE, CUST_CODE=CUST_CODE);

gg.handlerlist=filewriter

#The File Writer Handler
gg.handler.filewriter.type=filewriter
gg.handler.filewriter.mode=op
gg.handler.filewriter.pathMappingTemplate=./dirout
gg.handler.filewriter.stateFileDirectory=./dirsta
gg.handler.filewriter.fileNameMappingTemplate=${fullyQualifiedTableName}_${currentTimestamp}.txt
gg.handler.filewriter.fileRollInterval=7m
gg.handler.filewriter.finalizeAction=none
gg.handler.filewriter.inactivityRollInterval=7m
gg.handler.filewriter.format=existing_avro_schema_ocf
gg.handler.filewriter.includetokens=true
gg.handler.filewriter.partitionByTable=true
gg.handler.filewriter.rollOnShutdown=true

gg.format.timestamp=yyyy-MM-dd HH:mm:ss.SSS

#Avro Metadata provider must be enabled.
gg.mdp.type=avro
gg.mdp.schemaFilesPath=/scratch/tbcampbe/ggwork/gg23.1.0.0.0WORKING5/avromdp

  "type" : "record",
  "name" : "TCUSTMERNEST",
  "namespace" : "QASOURCE",
  "fields" : [ {
    "name" : "CUST_CODE",
    "type" : [ "null", "string" ],
    "default" : null,
    "primary_key": true
  }, {
    "name" : "SUB",
    "type" : {
      "type" : "record",
      "name" : "subrecord",
      "namespace" : "oracle.goldengate",
      "fields" : [ {
        "name" : "NAME",
        "type" : [ "null", "string" ],
        "default" : null
      }, {
        "name" : "STATE",
        "type" : [ "null", "string" ],
        "default" : null
      }, {
        "name" : "CITY",
        "type" : [ "null", "string" ],
        "default" : null
      } ]
    }
  } ]
}

To output the metacolumns configure the following:

To also include the primary key columns and the tokens configure as follows:

gg.handler.name.format.metaColumnsTemplate=${objectname[table]},${optype[op_type]},${timestamp[op_ts]},${currenttimestamp[current_ts]},${position[pos]},${primarykeycolumns[primary_keys]},${alltokens[tokens]}

For more information see the configuration property: gg.handler.name.format.metaColumnsTemplate.

JSON messages begin with the operation metadata fields, which are followed by the operation data fields. This data is represented by before and after members that are objects. These objects contain members whose keys are the column names and whose values are the column values.

Operation data is modeled as follows:

• Inserts: Includes the after-image data.

• Updates: Includes both the before-image and the after-image data.

• Deletes: Includes the before-image data.

Column values for an operation from the source trail file can have one of three states: the column has a value, the column value is null, or the column value is missing. The JSON Formatter maps these column value states into the created JSON objects as follows:

• The column has a value: The column value is output. In the following example, the member STATE has a value.

      "after":{        "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",        "STATE":"CO"    }
  

• The column value is null: The default output value is a JSON NULL. In the following example, the member STATE is null.

      "after":{        "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",        "STATE":null    }
  

• The column value is missing: The JSON contains no element for a missing column value. In the following example, the member STATE is missing.

      "after":{        "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",    }
  

The default setting of the JSON Formatter is to map the data types from the source trail file to the associated JSON data type. JSON supports few data types, so this functionality usually results in the mapping of numeric fields from the source trail file to members typed as numbers. This data type mapping can be configured treat all data as strings.

JSON messages begin with the operation metadata fields, which are followed by the operation data fields. For row data formatting, this are the source column names and source column values as JSON key value pairs. This data is represented by before and after members that are objects. These objects contain members whose keys are the column names and whose values are the column values.

Row data is modeled as follows:

• Inserts: Includes the after-image data.

• Updates: Includes the after-image data.

• Deletes: Includes the before-image data.

Column values for an operation from the source trail file can have one of three states: the column has a value, the column value is null, or the column value is missing. The JSON Formatter maps these column value states into the created JSON objects as follows:

• The column has a value: The column value is output. In the following example, the member STATE has a value.

          "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",        "STATE":"CO"    }
  

• The column value is null :The default output value is a JSON NULL. In the following example, the member STATE is null.

          "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",        "STATE":null    }
  

• The column value is missing: The JSON contains no element for a missing column value. In the following example, the member STATE is missing.

          "CUST_CODE":"BILL",        "NAME":"BILL'S USED CARS",        "CITY":"DENVER",    }
  

The default setting of the JSON Formatter is to map the data types from the source trail file to the associated JSON data type. JSON supports few data types, so this functionality usually results in the mapping of numeric fields from the source trail file to members typed as numbers. This data type mapping can be configured to treat all data as strings.

The following topics are sample JSON messages created by the JSON Formatter for insert, update, delete, and truncate operations.

By default, JSON schemas are generated for each source table encountered. JSON schemas are generated on a just in time basis when an operation for that table is first encountered. Newer schemas are generated when there is a change in the metadata. A JSON schema is not required to parse a JSON object. However, many JSON parsers can use a JSON schema to perform a validating parse of a JSON object. Alternatively, you can review the JSON schemas to understand the layout of output JSON objects. By default, the JSON schemas are created in the GoldenGate_Home/dirdef directory and are named by the following convention:

FULLY_QUALIFIED_TABLE_NAME.schema.json

The generation of the JSON schemas is suppressible.

The following JSON schema example is for the JSON object listed in Sample Operation Modeled JSON Messages.

{
    "$schema":"http://json-schema.org/draft-04/schema#",
    "title":"QASOURCE.TCUSTORD",
    "description":"JSON schema for table QASOURCE.TCUSTORD",
    "definitions":{
        "row":{
            "type":"object",
            "properties":{
                "CUST_CODE":{
                    "type":[
                        "string",
                        "null"
                    ]
                },
                "ORDER_DATE":{
                    "type":[
                        "string",
                        "null"
                    ]
                },
                "PRODUCT_CODE":{
                    "type":[
                        "string",
                        "null"
                    ]
                },
                "ORDER_ID":{
                    "type":[
                        "number",
                        "null"
                    ]
                },
                "PRODUCT_PRICE":{
                    "type":[
                        "number",
                        "null"
                    ]
                },
                "PRODUCT_AMOUNT":{
                    "type":[
                        "integer",
                        "null"
                    ]
                },
                "TRANSACTION_ID":{
                    "type":[
                        "number",
                        "null"
                    ]
                }
            },
            "additionalProperties":false
        },
        "tokens":{
            "type":"object",
            "description":"Token keys and values are free form key value pairs.",
            "properties":{
            },
            "additionalProperties":true
        }
    },
    "type":"object",
    "properties":{
        "table":{
            "description":"The fully qualified table name",
            "type":"string"
        },
        "op_type":{
            "description":"The operation type",
            "type":"string"
        },
        "op_ts":{
            "description":"The operation timestamp",
            "type":"string"
        },
        "current_ts":{
            "description":"The current processing timestamp",
            "type":"string"
        },
        "pos":{
            "description":"The position of the operation in the data source",
            "type":"string"
        },
        "primary_keys":{
            "description":"Array of the primary key column names.",
            "type":"array",
            "items":{
                "type":"string"
            },
            "minItems":0,
            "uniqueItems":true
        },
        "tokens":{
            "$ref":"#/definitions/tokens"
        },
        "before":{
            "$ref":"#/definitions/row"
        },
        "after":{
            "$ref":"#/definitions/row"
        }
    },
    "required":[
        "table",
        "op_type",
        "op_ts",
        "current_ts",
        "pos"
    ],
    "additionalProperties":false
}

The following JSON schema example is for the JSON object listed in Sample Flattened Operation Modeled JSON Messages.

{
    "$schema":"http://json-schema.org/draft-04/schema#",
    "title":"QASOURCE.TCUSTORD",
    "description":"JSON schema for table QASOURCE.TCUSTORD",
    "definitions":{
        "tokens":{
            "type":"object",
            "description":"Token keys and values are free form key value pairs.",
            "properties":{
            },
            "additionalProperties":true
        }
    },
    "type":"object",
    "properties":{
        "table":{
            "description":"The fully qualified table name",
            "type":"string"
        },
        "op_type":{
            "description":"The operation type",
            "type":"string"
        },
        "op_ts":{
            "description":"The operation timestamp",
            "type":"string"
        },
        "current_ts":{
            "description":"The current processing timestamp",
            "type":"string"
        },
        "pos":{
            "description":"The position of the operation in the data source",
            "type":"string"
        },
        "primary_keys":{
            "description":"Array of the primary key column names.",
            "type":"array",
            "items":{
                "type":"string"
            },
            "minItems":0,
            "uniqueItems":true
        },
        "tokens":{
            "$ref":"#/definitions/tokens"
        },
        "before.CUST_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "before.ORDER_DATE":{
            "type":[
                "string",
                "null"
            ]
        },
        "before.PRODUCT_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "before.ORDER_ID":{
            "type":[
                "number",
                "null"
            ]
        },
        "before.PRODUCT_PRICE":{
            "type":[
                "number",
                "null"
            ]
        },
        "before.PRODUCT_AMOUNT":{
            "type":[
                "integer",
                "null"
            ]
        },
        "before.TRANSACTION_ID":{
            "type":[
                "number",
                "null"
            ]
        },
        "after.CUST_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "after.ORDER_DATE":{
            "type":[
                "string",
                "null"
            ]
        },
        "after.PRODUCT_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "after.ORDER_ID":{
            "type":[
                "number",
                "null"
            ]
        },
        "after.PRODUCT_PRICE":{
            "type":[
                "number",
                "null"
            ]
        },
        "after.PRODUCT_AMOUNT":{
            "type":[
                "integer",
                "null"
            ]
        },
        "after.TRANSACTION_ID":{
            "type":[
                "number",
                "null"
            ]
        }
    },
    "required":[
        "table",
        "op_type",
        "op_ts",
        "current_ts",
        "pos"
    ],
    "additionalProperties":false
}

The following JSON schema example is for the JSON object listed in Sample Row Modeled JSON Messages.

{
    "$schema":"http://json-schema.org/draft-04/schema#",
    "title":"QASOURCE.TCUSTORD",
    "description":"JSON schema for table QASOURCE.TCUSTORD",
    "definitions":{
        "tokens":{
            "type":"object",
            "description":"Token keys and values are free form key value pairs.",
            "properties":{
            },
            "additionalProperties":true
        }
    },
    "type":"object",
    "properties":{
        "table":{
            "description":"The fully qualified table name",
            "type":"string"
        },
        "op_type":{
            "description":"The operation type",
            "type":"string"
        },
        "op_ts":{
            "description":"The operation timestamp",
            "type":"string"
        },
        "current_ts":{
            "description":"The current processing timestamp",
            "type":"string"
        },
        "pos":{
            "description":"The position of the operation in the data source",
            "type":"string"
        },
        "primary_keys":{
            "description":"Array of the primary key column names.",
            "type":"array",
            "items":{
                "type":"string"
            },
            "minItems":0,
            "uniqueItems":true
        },
        "tokens":{
            "$ref":"#/definitions/tokens"
        },
        "CUST_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "ORDER_DATE":{
            "type":[
                "string",
                "null"
            ]
        },
        "PRODUCT_CODE":{
            "type":[
                "string",
                "null"
            ]
        },
        "ORDER_ID":{
            "type":[
                "number",
                "null"
            ]
        },
        "PRODUCT_PRICE":{
            "type":[
                "number",
                "null"
            ]
        },
        "PRODUCT_AMOUNT":{
            "type":[
                "integer",
                "null"
            ]
        },
        "TRANSACTION_ID":{
            "type":[
                "number",
                "null"
            ]
        }
    },
    "required":[
        "table",
        "op_type",
        "op_ts",
        "current_ts",
        "pos"
    ],
    "additionalProperties":false
}

The following is a sample configuration for the JSON Formatter in the Java Adapter configuration file:

gg.handler.hdfs.format=json
gg.handler.hdfs.format.insertOpKey=I
gg.handler.hdfs.format.updateOpKey=U
gg.handler.hdfs.format.deleteOpKey=D
gg.handler.hdfs.format.truncateOpKey=T
gg.handler.hdfs.format.prettyPrint=false
gg.handler.hdfs.format.jsonDelimiter=CDATA[]
gg.handler.hdfs.format.generateSchema=true
gg.handler.hdfs.format.schemaDirectory=dirdef
gg.handler.hdfs.format.treatAllColumnsAsStrings=false

Metadata change events are handled at runtime. When metadata is changed in a table, the JSON schema is regenerated the next time an operation for the table is encountered. The content of created JSON messages changes to reflect the metadata change. For example, if an additional column is added, the new column is included in created JSON messages after the metadata change event.

When the JSON formatter is configured to model operation data, primary key updates require no special treatment and are treated like any other update. The before and after values reflect the change in the primary key.

When the JSON formatter is configured to model row data, primary key updates must be specially handled. The default behavior is to abend. However, by using thegg.handler.name.format.pkUpdateHandling configuration property, you can configure the JSON formatter to model row data to treat primary key updates as either a regular update or as delete and then insert operations. When you configure the formatter to handle primary key updates as delete and insert operations, Oracle recommends that you configure your replication stream to contain the complete before-image and after-image data for updates. Otherwise, the generated insert operation for a primary key update will be missing data for fields that did not change.

You can integrate Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) with Oracle Stream Analytics (OSA) by sending operation-modeled JSON messages to the Kafka Handler. This works only when the JSON formatter is configured to output operation-modeled JSON messages.

Because OSA requires flattened JSON objects, a new feature in the JSON formatter generates flattened JSONs. To use this feature, set the gg.handler.name.format.flatten=false to true. (The default setting is false). The following is an example of a flattened JSON file:

{
    "table":"QASOURCE.TCUSTMER",
    "op_type":"U",
    "op_ts":"2015-11-05 18:45:39.000000",
    "current_ts":"2016-06-22T13:38:45.335001",
    "pos":"00000000000000005100",
    "before.CUST_CODE":"ANN",
    "before.NAME":"ANN'S BOATS",
    "before.CITY":"SEATTLE",
    "before.STATE":"WA",
    "after.CUST_CODE":"ANN",
    "after.CITY":"NEW YORK",
    "after.STATE":"NY"
}

• Column 0 as "_id", which identifies a document in a collection.
• Column 1 as "payload", which holds all the columns (fields of a collection).

JSON Mongo Document Formatter formats the MongoDB Capture processed documents into a JSON format with only payload information.

Example

The document from trail received is

{"after":{"id":"{ \"_id\" :

{ \"$oid\" : \"65b9f02b80f1c27eb4b498e1\" }

}", "payload":"{\"_id\":

{\"$oid\": \"65b9f02b80f1c27eb4b498e1\"}

, \"CUST_CODE\":

\"test2\", \"name\": \"hello world\", \"cost\": {\"$numberDouble\": \"3000.0\"}}"}}

{"data":"{\"_id\":

{\"$oid\": \"65b9f02b80f1c27eb4b498e1\"}

, \"CUST_CODE\": \"test2\", \"name\": \"hello world\", \"cost\": {\"$numberDouble\": \"3000.0\"}}"}

JSON MongoDocument Formatter can be configured to write the data either in JSON EXTENDED format or JSON RELAXED format with payload value.

Required Dependencies

Oracle GoldenGate requires that you use the 4.11.1 bson library with JSON Mongo Document Formatter. You can download this driver from: https://mvnrepository.com/artifact/org.mongodb/bson/4.11.1

Maven artifacts for bson-4.11.1 as follows:

<dependency>

                <groupId>org.mongodb</groupId>

                <artifactId>bson</artifactId>

                <version>4.11.1</version>

           </dependency>

You must include the path of the bson library in the gg.classpath property.

Example:

gg.classpath=./bson-4.11.1.jar

JSON MongoDocument Formatter Configuration Properties

The following is a sample configuration for the JSON Mongo Document Formatter in the Java Adapter configuration file:

gg.handler.kafka.format=mongodocument

gg.handler.kafka.format.insertOpKey=I

gg.handler.kafka.format.updateOpKey=U

gg.handler.kafka.format.deleteOpKey=D

gg.handler.kafka.format.truncateOpKey=T

gg.handler.kafka.format.metaColumnsTemplate=${optype},${timestampmicro},${currenttimestampmicro},${timestamp}

The default format for output of data is the following:

First is the row Length followed by metadata:

<ROW LENGTH><PRESENT INDICATOR><FIELD LENGTH><OPERATION TYPE><PRESENT INDICATOR><FIELD LENGTH><FULLY QUALIFIED TABLE NAME><PRESENT INDICATOR><FIELD LENGTH><OPERATION TIMESTAMP><PRESENT INDICATOR><FIELD LENGTH><CURRENT TIMESTAMP><PRESENT INDICATOR><FIELD LENGTH><TRAIL POSITION><PRESENT INDICATOR><FIELD LENGTH><TOKENS>

Or

<ROW LENGTH><FIELD LENGTH><FULLY QUALIFIED TABLE NAME><FIELD LENGTH><OPERATION TIMESTAMP><FIELD LENGTH><CURRENT TIMESTAMP><FIELD LENGTH><TRAIL POSITION><FIELD LENGTH><TOKENS>	

Next is the row data:

<PRESENT INDICATOR><FIELD LENGTH><COLUMN 1 VALUE><PRESENT INDICATOR><FIELD LENGTH><COLUMN N VALUE>

Insert Message:

0133P01IP161446749136000000P161529311765024000P262015-11-05 
18:45:36.000000P04WILLP191994-09-30 15:33:00P03CARP03144P0817520.00P013P03100

Update Message

0133P01UP161446749139000000P161529311765035000P262015-11-05 
18:45:39.000000P04BILLP191995-12-31 15:00:00P03CARP03765P0814000.00P013P03100

Delete Message

0136P01DP161446749139000000P161529311765038000P262015-11-05 
18:45:39.000000P04DAVEP191993-11-03 
07:51:35P05PLANEP03600P09135000.00P012P03200

${optype}, ${token.ROWID}, ${sys.username}, ${currenttimestamp}

Review a Sample Configuration

#The LDV Handler
gg.handler.filewriter.format=binary
gg.handler.filewriter.format.binaryLengthMode=false
gg.handler.filewriter.format.recordLength=4
gg.handler.filewriter.format.fieldLength=2
gg.handler.filewriter.format.legacyFormat=false
gg.handler.filewriter.format.presentValue=CDATA[P]
gg.handler.filewriter.format.missingValue=CDATA[M]
gg.handler.filewriter.format.nullValue=CDATA[N]
gg.handler.filewriter.format.metaColumnsTemplate=${optype},${timestampmicro},${currenttimestampmicro},${timestamp}
gg.handler.filewriter.format.pkUpdateHandling=abend

Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) differs from RDBMSs in how data is stored. Update and delete operations in an RDBMS result in a change to the existing data. Data is not changed in GG for DAA, it is simply appended to existing data. The current state of a given row becomes a consolidation of all of the existing operations for that row in the HDFS system.

Primary Key Updates

Consolidating Data

GG for DAA simply append data to the underlying storage. Typically, analytic tools spawn map reduce programs that traverse the data files and consolidate all the operations for a given row into a single output. It is important to have an indicator of the order of operations. The Length Delimited Value Formatter provides a number of metadata fields to fulfill this need. The operation timestamp may be sufficient to fulfill this requirement. However, two update operations may have the same operation timestamp especially if they share a common transaction. The trail position can provide a tie breaking field on the operation timestamp. Lastly, the current timestamp may provide the best indicator of order of operations in GG for DAA.

The XML formatted messages contain the following information:

The following sections provide sample XML messages.

The XML Formatter does not generate an XML schema (XSD). The XSD applies to all messages generated by the XML Formatter. The following XSD defines the structure of the XML documents that are generated by the XML Formatter.

<xs:schema attributeFormDefault="unqualified" 
elementFormDefault="qualified" xmlns:xs="http://www.w3.org/2001/XMLSchema">
   <xs:element name="operation">
     <xs:complexType>
       <xs:sequence>
          <xs:element name="col" maxOccurs="unbounded" minOccurs="0">
           <xs:complexType>
             <xs:sequence>
               <xs:element name="before" minOccurs="0">
                 <xs:complexType>
                   <xs:simpleContent>
                     <xs:extension base="xs:string">
                       <xs:attribute type="xs:string" name="missing" 
use="optional"/>
                     </xs:extension>
                   </xs:simpleContent>
                 </xs:complexType>
               </xs:element>
               <xs:element name="after" minOccurs="0">
                 <xs:complexType>
                   <xs:simpleContent>
                     <xs:extension base="xs:string">
                       <xs:attribute type="xs:string" name="missing" 
use="optional"/>
                     </xs:extension>
                   </xs:simpleContent>
                 </xs:complexType>
               </xs:element>
               <xs:element type="xs:string" name="missing" minOccurs="0"/>
             </xs:sequence>
             <xs:attribute type="xs:string" name="name"/>
             <xs:attribute type="xs:short" name="index"/>
           </xs:complexType>
         </xs:element>
         <xs:element name="tokens" minOccurs="0">
           <xs:complexType>
             <xs:sequence>
               <xs:element name="token" maxOccurs="unbounded" minOccurs="0">
                 <xs:complexType>
                   <xs:sequence>
                     <xs:element type="xs:string" name="Name"/>
                     <xs:element type="xs:string" name="Value"/>
                   </xs:sequence>
                 </xs:complexType>
               </xs:element>
             </xs:sequence>
           </xs:complexType>
         </xs:element>
       </xs:sequence>
       <xs:attribute type="xs:string" name="table"/>
       <xs:attribute type="xs:string" name="type"/>
       <xs:attribute type="xs:string" name="ts"/>
       <xs:attribute type="xs:dateTime" name="current_ts"/>
       <xs:attribute type="xs:long" name="pos"/>
       <xs:attribute type="xs:short" name="numCols"/>
     </xs:complexType>
   </xs:element>
</xs:schema>

The following is a sample configuration for the XML Formatter in the Java Adapter properties file:

gg.handler.hdfs.format=xml
gg.handler.hdfs.format.insertOpKey=I
gg.handler.hdfs.format.updateOpKey=U
gg.handler.hdfs.format.deleteOpKey=D
gg.handler.hdfs.format.truncateOpKey=T
gg.handler.hdfs.format.encoding=ISO-8859-1
gg.handler.hdfs.format.includeProlog=false

The XML Formatter seamlessly handles metadata change events. A metadata change event does not result in a change to the XML schema. The XML schema is designed to be generic so that the same schema represents the data of any operation from any table.

If the replicated database and upstream Oracle GoldenGate replication process can propagate metadata change events, the XML Formatter can take action when metadata changes. Changes in the metadata are reflected in messages after the change. For example, when a column is added, the new column data appears in XML messages for the table.

Updates to a primary key require no special handling by the XML formatter. The XML formatter creates messages that model database operations. For update operations, this includes before and after images of column values. Primary key changes are represented in this format as a change to a column value just like a change to any other column value.

The FW handler needs to be chained to an object store Event handler that can upload the staging files into a staging location.

The staging location is typically a cloud object store, such as AWS S3 or Azure Data Lake.

In most cases FW handler is configured to use the Avro Object Container Format (OCF) formatter.

The output file format could change based on the specific data warehouse target.

Operation Aggregation needs to be enabled for stage and merge replication using the configuration gg.aggregate.operations=true.

• replicat parameter (.prm) file.
• replicat properties file that contains the FW handler and all the Event handler configuration.
• DDL file for the sample table used in the merge script.
• Merge script for the specific data warehouse. This script contains SQL statements tested using the sample table defined in the DDL file.

#TODO: Edit this. Provide the replicat group name.
repName=RBD

#TODO: Edit this. Ensure each replicat uses a unique prefix.
stagingTablePrefix=${repName}_STAGE_

#TODO: Edit the AWS S3 bucket name.
bucket=<AWS S3 bucket name>

#TODO: Edit this variable as needed.
s3Location="'s3://${bucket}/${dir}/'"

#TODO: Edit AWS credentials awsKeyId and awsSecretKey
awsKeyId=<AWS Access Key Id>
awsSecretKey=<AWS Secret key>

The variables repName and stagingTablePrefix are relevant for all the data warehouse targets.

In most cases, we need to double quote " identifiers in the SQL statement. The double quote needs to be escaped in the script using backslash. For example: \".

Oracle provides a working example of SQL statements for a single table with a pre-defined set of columns defined in the sample DDL file. You need to add new sections for your own tables as part of if-else code block in the script.

if [ "${tableName}" == "DBO.TCUSTORD" ]
then
  #TODO: Edit all the column names of the staging and target tables.
  # The merge SQL example here is configured for the example table defined in the DDL file.
  # Oracle provided SQL statements

# TODO: Add similar SQL queries for each table.
elif [ "${tableName}" == "DBO.ANOTHER_TABLE" ]
then
  
#Edit SQLs for this table.
fi

The script is coded to include the following shell functions:

• main
• validateParams
• process
• processTruncate
• processDML
• dropExternalTable
• createExternalTable
• merge

The script has code comments for you to infer the purpose of each function.

Merge Script main function

The function main is the entry point of the script. The processing of the staged changed data file begin here.

This function invokes two functions: validateParams and process.

The input parameters to the script is validated in the function: validateParams.

Processing resumes in the process function if validation is successful.

Merge Script process function

This function processes the operation records in the staged change data file and invokes processTruncate or processDML as needed.

Truncate operation records are handled in the function processTruncate. Insert, Update, and Delete operation records are handled in the function processDML.

Merge Script merge function

The merge function invoked by the function processDML contains the merge SQL statement that will be executed for each table.

The key columns to be used in the merge SQL’s ON clause needs to be customized.

ON ((NVL(CAST(TARGET.\"C01Key\" AS VARCHAR(4000)),'${uuid}')=NVL(CAST(STAGE.\"C01Key\" AS VARCHAR(4000)),'${uuid}')))`

The column names in the merge statement’s update and insert clauses also needs to be customized for every table.

Merge Script createExternalTable function

The createExternalTable function invoked by the function processDML creates an external table that is backed by the file in the respective object store file.

In this function, the DDL SQL statement for the external table should be customized for every target table to include all the target table columns.

In addition to the target table columns, the external table definition also consists of three meta-columns: optype, position, and fieldmask.

The data type of the meta-columns should not be modified. The position of the meta-columns should not be modified in the DDL statement.

• The Command handler merge scripts are available, starting from Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) release 19.1.0.0.8.
• The respective data warehouse’s command line programs to execute SQL queries must be installed on the machine where GG for DAA is installed.

Primary key update operations are split into delete and insert pair. In case the Oracle GoldenGate trail file doesn't contain column values for all the columns in the respective table, then the missing columns gets updated to null on the target table.