Push Data into Celonis with the Data Ingestion API

It's time to push data! In this course, you'll learn how to push data using Celonis' Data Ingestion API.

Learning Objectives

• Understand the Data Ingestion API's architectural foundation and specifications
• Understand which use cases apply for the Data Ingestion API
• Perform a data push using the Data Ingestion API in the ML Workbench

Prerequisites

We recommend you go through the first milestone of the Get Data into Celonis track to understand the basics of Data Integration. Also you should have a basic understanding of APIs and Python.

Course Duration

~1 hour

Personal Training Environment Required

This course contains hands-on exercises that require a personal Celonis Training Environment, also called a "Celonis Team".

Not sure if you have one? Click below and we'll either create one for you or show you how to access your existing Team if you have one. Please make sure to disable adblockers on this page if the button is not working for you.

Check / Create my training environment

Let's get started!

Tip: You can Hide the Menu bar (at the top left).

---

Architecture

The Data Ingestion API relies on an ingestion architecture separate from the native extractors.

This ingestion area is built on top of cloud native object storages (S3) and can handle massive data volumes. As it arrives data is immediately unnested and pushed to Data Storage (Vertica) and ready for processing. Here is a example flow with a Kafka system:

For more information on object storage and its benefits, refer to AWS' object storage definition.

Specifications

Here are a few things you should know about the API itself:

One API call per table: Every table you create or update requires one API call. Built on S3 API: It's built on top of the S3 API with the same methods and error codes. It uses primarily this S3 PUT object endpoint. Continuous load: The load is continuous. So as soon as you ping the API, your data is loaded into Celonis. It operates on a First In First Out (FIFO) principle. Auto-delta load: All pushed records are loaded as delta records automatically and compared to existing records. Nested data possible: The API can handle nested (json) data out-of-the-box and unnest it into tables and columns. UI schema configuration: It supports a UI based configuration for table schema, i.e. table names, keys, age columns. Parquet files only: As of today, the API only supports parquet files and requires you to convert to parquet before pushes.

Now that we've covered the basics, let's move on to use cases.

---

Two General Use Cases

There are two general use cases where pushing data is most likely a better fit than pulling data in Celonis:

Existing data applications: data resides or is processed through other data applications such as: ETL tools (e.g. Informatica) Streaming Applications (e.g. Kafka) IPaaS (e.g. Mulesoft) Other scripts or Action Flows that rely on the older Data Push API or Continuous Data Push API Ongoing file pushes Security concerns: data needs to be pushed into Celonis for security reasons, e.g. access to the underlying source cannot be granted.

In other words, if data should not or cannot be extracted directly from source systems, you should consider the Data Ingestion API.

System Type Specifics

Depending on the existing data application, you'll need to consider a few extra aspects. Below is an overview of these aspects per system type.

SYSTEM TYPE FACTORS TO CONSIDER ETL Tools Informatica is a good example for this category. With no existing extractors for these, the Data Ingestion API is the right option. Streaming Applications Kafka is the main candidate in this category. The setup for this requires no coding as the Kafka Connector configuration can be exported directly from the connection details and used as configuration to start your Kafka Connector.

IPaaS Integration Platform as a Service systems such as Mulesoft often serve as a integration layer to get data from system A to system B. This is a good use case to use the Data Ingestion API. Cloud data lakes/warehouses  These refer mostly to the modern databases such as Snowflake, Databricks, BigQuery, Redshift or Azure Synapse. For these, Celonis offers "pull" extractors either as standalone or through the JDBC/Database extractor. The decision here comes down to the following: Extraction filters, column removals, data types changes, data integration scheduling? → JDBC Database extractor Low latency needed? → Data Ingestion API Large data volumes? → Data Ingestion API Security concerns? → Data Ingestion API (offers full control on what data leaves the premises)

If you plan using the Data Ingestion API, then look for the providers' documentation on how to externally Create a Table:

Redshift Databricks BigQuery Snowflake

Scripts and Actions Flows based on older APIs

Before the Data Ingestion API, the Data Push API and Continous Data Push API were used together with scripts to setup a data "push" method.

Due to the complexity of the setup, all these scenarios should should go to the much simpler Data Ingestion API.

In case of Action Flows, users can also use the native Celonis module 'Write back to data pool table' rather than developing a custom HTTP module.

Ongoing file pushes Many projects initially start with file pushes for a quick data check. This initial push is normally done using the file uploader. That said if pushing files (parquet) should be the long-term approach, then the Data Ingestion API is the right way to go.

Enough theory. Next up, sit back and watch our product manager's demo.

---

Watch this video from our Product Manager Florian Sonneck. It covers:

A recap of the Data Ingestion API theoretical foundations A sample flat table push of a parquet file via Python using the Celonis Machine Learning Workbench A sample nested table push using the native AWS UI How to work with primary and foreign keys

Note: Don't worry about following along, this demo is quite fast. After watching this, it's your turn but you'll get all the steps!

_Media:_

• https://fast.wistia.net/embed/iframe/0jvavqr2wx?videoFoam=true

---

Did you watch the video on the previous page? Let's push a flat parquet file using the Celonis ML workbench. You can challenge yourself with just the basic steps or follow the step-by-step instructions.

Basic Steps

There are three high-level steps here:

Create the data pool and ingestion connection. Call the FLAT table "Orders" and use order_id as the Primary Key. Run the API call in ML Workbench to push the flat_order.parquet (download the files in Resources) You'll need to create a new app. (This may take 2-3 minutes). And run the Ingestion API single file.ipynb file, adjusting it's values. If you're missing boto3, you can install it in a terminal using "python -m pip install boto3" or "python -m pip install boto3 --upgrade" if an upgrade is needed Check your data Step by Step

There are three high-level steps here:

Create the data pool and ingestion connection Run the API call in ML Workbench to push the flat_order.parquet (download the files in Resources) You'll need to create a new app. And run the Ingestion API single file.ipynb file, adjusting it's values. If you're missing boto3, you can install it in a terminal using "python -m pip install boto3" or "python -m pip install boto3 --upgrade" if an upgrade is needed Check your data Create the pool and connection Navigate to your training environment by logging in to Celonis ID and selecting your team under Celonis Academy. In Data Integration, create a new data pool or select an existing one, and create a new connection: Select "Push data into Celonis": Give your connection a name and then add the source table and target table as "Orders" with "order_id" as the primary key. Did you notice the "Age columns" field? This field is typically a date column that indicates when a record was created or updated. It works together with the primary key to identify delta records (updates / new records). Save and copy your access key and access secret for later. After closing the pop-up, select your connection and go so S3 Clients configuration in the vertical three dot menu. Go to the Python tab and copy the python code to your clipboard. You will use this code in the ML workbench. Run the API call in ML Workbench In a separate tab, navigate to Data → Machine Learning. Select the Unassigned apps workspace, then click "Create Workbench" on the top right, give it a name of your choice, and click "Create". (This may take 2-3 minutes to load) Start the Jupyter Lab Download this zip of sample files, and unzip it. Select your new ML app and drag and drop the files unzipped files into the File Explorer area: Start a new Python Notebook and paste in the python code you copied from the S3 client configurations on your connection.

Replace and ensure you have the following values: Access Key Access Secret Endpoint URL (replace everything up to "cloud" in the URL with your Celonis teams URL) The parquet_file path to 'flat_order.parquet'. The s3_object_name to 'flat_order.parquet'. Run the code by pressing the play icon. In case boto3 is not installed (you see an error), create a new command line session and run the following command without the double quotes: "python -m pip install boto3"

If an upgrade is needed, you can run this command without the double quotes:

"python -m pip install boto3 --upgrade" After the installation runs successfully. Go back to your Python notebook tab. Run each cell using the play button or use "Run all Cells" in the Run dropdown menu. If boto3 is not recognized, try restarting the kernel in the Kernel menu. You should see this message at the bottom if it ran successfully: Check your data Go to your data pool and create a new data job selecting your data ingestion connection. Create a new transformation and run a SELECT * FROM {your table name} to check the data.

That was it for the flat file via the ML Workbench. If you want a mini extra challenge, you can update this table's record by running the code for the flat_order2.parquet to update the existing table.

Next, let's work on a nested table using the AWS CLI.

---

Let's nowpush a nested table (parquet) using the AWS CLI. Challenge yourself using the basic steps or simply follow the step by step.

Basic Steps

The high-level steps for this exercise are:

Add the nested table to the connection - including its schema based on the nested_order.json file. Use AWS CLI in a terminal to push data - Install the AWS CLI using "pip install awscli". If it needs an upgrade or another library does, make sure to run e.g. "pip install --upgrade awscli" The commands you need are the Command Line.ipynb file. The region for the AWS profile if you're on a training team is eu-central-1. Check your data Step by Step

The high-level steps for this exercise are:

Add the nested table to the connection Use AWS CLI to push data Check your data Add the nested table to the connection Navigate to your ingestion connection and add a table called "OrderLines", set order_id as the primary key. Then navigate to schema configuration. Open the "nested_order.json" file (from the previous zip download) with a Notepad tool. Copy and paste the schema into the preview box on the left. Scroll down to the nested table and set the order_line_id as the nested table's primary key. Click "Finish" and save your new table.

As you saw, for nested tables you always need to define the schema using a json sample of the data.

Use AWS CLI to push data Before going to the ML Workbench, copy the list of commands for the AWS CLI again from the S3 Client Configurations on the connection: Navigate to your the ML Workbench App and open a new Terminal as well as the Command Line.ipynb file. In the Terminal tab, run this command to install the python aws cli package: pip install awscli

awscli or another library may need an update, so simply update it using e.g. pip install --upgrade awscli

Next in Terminal, create a profile by copying and pasting the first command from the code: aws configure --profile demo You'll need to enter: the Access Key the Access Secret the region of your realm: {you can leave this empty and press Enter} the output type : {you can leave this empty and press Enter} Then edit the code you copied adding in the name of your profile (e.g. demo), the path and name of your file in two areas (nested_order.parquet). Here is an adjusted sample: Once edited, copy it into the command line tool and run it.

But what was the command you just ran? Here are the parts: aws s3api put-object : This is the standard AWS API call to copy objects/ files to a S3 bucket nested_order.parquet: The name of the file getting copied to the S3 bucket --endpoint-url {Team URL/api/data-ingestion}: The S3 bucket is behind a Celonis specific url to relate it to the Celonis team and cluster (e.g. us-1 or training). --bucket continuous: The name of the target bucket. In the context of Celonis, this usually refers to a specific data pool or ingestion area. --key: This is the destination path and filename. The bucket's 32-digit id is your Celonis connection ID. And "OrderLines" is the name of the table you've configured into which data gets pushed. Since one API call pushes to one specific table and its nested table. You need one call per parent table. --body: This points to the local file on your computer that you want to upload. --profile profileName: This refers to the AWS profile you've configured in the previous step Check your data Go back to your transformation in your data pool. And check the results: Do you see how two tables were imported? The parent table "OrderLines" for the Orders and the nested table "OrderLines_line_items". In a sense, naming the parent table "OrderLines" here does not make sense. It should be simply "Orders" but we already had this name from the previous exercise. Alternatively, we could have re-worked the existing table into a nested table and push the data there.

That's it for the nested table push! The next page covers some questions you may have after going through this exercise.

That's it for the nested table push! The next page covers some questions you may have after going through this exercise.

---

Here are a few frequently asked questions on the Data Ingestion API.

How can I schedule the pushes?

Celonis extraction schedules are not possible when using the Data Ingestion API. Scheduling should be done in the tool hosting the API calls.

What happens if the table schemas change?

As long as the primary key remains the same, a push with new columns will simply add the columns to the existing table and enter Null values if an older column is excluded.

How do Delta Loads work?

Delta loads (updates or new records) rely on the Primary Key and the Age field. The Age field is typically a date column that indicates when a record was created or updated.

If a new record is inserted with the same Primary Key value, then the one with the latest date is retained. If no Age field is defined, then the new push's record is kept and updates the existing one.

Note that you can only define the Age column at the top table level. Nested tables rely solely on their primary keys and the parent table's fields for delta logic.

How do I convert my files to parquet?

In the sample files we provided, the files were already converted. Here you simple need to look for a package / library that can do it for you. For python for example you can use

pyarrow for example:

from pyarrow import json import pyarrow.parquet as pq

table = json.read_json('C:/python/json_file') pq.write_table(table, 'C:/python/result.parquet') # save json/table as parquet What happens if I change my table's primary key?

Changing a table's primary key means delta loads won't be identified correctly. In this case, we recommend dropping your existing table and performing a full load unless you want to keep the old records.

Similarly, if you remove the primary key, the load will go into "append" mode and no longer check if there are duplicate records.

How can I delete my tables?

For a full reset, we recommend you delete both the ingestion connection table in the UI and the loaded table within your schema.

If there is a table dependency, you may have to add a "CASCADE" to your DROP TABLE SQL statement:

DROP TABLE {table_name} CASCADE How can I push many files from a folder?

This requires a few extra lines of coding. In the resources of this course, you'll find a sample "Ingestion API entire folder". This sample code pushes multiple files to the same table.

How can I monitor and fix errors?

Once a file transfer to Celonis is successfully, it may still throw errors in the ingestion. To see these errors, you can go to the Monitoring option on the connection.

For certain issues you can fix the table schema and retry the ingestion. For others you may have the fix is in the source file itself. Check our documentation on typical ingestion errors for more information.

---

Congratulations on completing this course and mastering the art of the push!

Here is a summary of what we covered:

The Data Ingestion API's architectural foundation and basic specifications. The two major use cases that apply for the Data Ingestion API - existing data systems or security concerns. How to perform a data push using the Data Ingestion API in the ML Workbench via python or a terminal and the AWS CLI tool.

If at any time you need a reference for the API, we invite you to go to our developer documentation.

Remember to visit the Feedback page. Your feedback is optional but greatly appreciated.

Want to dig deeper into the topic of this course? Join the Celonis Community to ask your product questions, hear about the latest product releases, and remain up-to-date by subscribing.

We invite you to join our Celonis Academy Group to engage with your peers, get exclusive updates and answers directly from Academy experts, and stay connected!