Amazon S3 (Simple Storage Service) is an AWS service built for scalable file storage. The S3 data integration to the Data Warehouse created by Kondado allows you to access CSV files in your analytical cloud.

Adding the connector

Consider that your bucket is called generic-bucket-name

Step 1: Create a New IAM Policy

1. Go to the IAM Console: https://console.aws.amazon.com/iam
2. In the left menu, click Policies.
3. Click the Create policy button.
4. In the JSON tab, paste the following policy document
   1. { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::generic-bucket-name", "arn:aws:s3:::generic-bucket-name/*" ] } ] }
5. Click Next: Tags, then Next: Review.
6. In the Policy name field, enter S3ReadListBucketGenericPolicy.
7. Click Create policy.

Step 2: Create a New IAM User

1. Go back to the IAM Console: https://console.aws.amazon.com/iam
2. In the left menu, click Users.
3. Click Add user.
4. Enter the username as S3ReadUserGenericBucket.
5. Select Access key - Programmatic access.
6. Click Next: Permissions.

Step 3: Attach the Policy to the User

1. On the Permissions page, click Attach existing policies directly.
2. Search for the policy you created: S3ReadListBucketGenericPolicy.
3. Check the box next to the policy and click Next: Tags, then Next: Review.
4. Click Create user.

Step 4: Download the Access Keys

1. After creating the user, the Access key ID and Secret access key will be shown.
2. Download or copy these credentials, as they will not be shown again.

The S3ReadUserGenericBucket user now has programmatic access to read and list files from the generic S3 bucket.

Step 5: Add on Kondado

1. On the Kondado platform, click + Create > Connector > select the S3 connector
2. Give your connector a name, fill in the values obtained in step 4 and the name of the bucket (in our example generic-bucket-name) where your files are located.

Now just save the connector and test the connection

Integrations

CSV

The integration for reading CSV files will create a table in your destination where all fields will be of type text and treated to replace special characters

All files must have a header

The following parameters are available:

1. Start reading date: Refers to the modification date of the files. Indicates from which date the data will start being read. If you choose to leave your integration as Full, this parameter will be ignored
2. Column delimiter: Indicate which character is used to separate the columns in the file
3. File prefix: Indicate the prefix of the files that will be included; do not start with the bucket name or with "/" or "s3://". Do not use wildcards. Example: folder_x/folder_y/start_of_the_file_name_
4. Compression: Choose GZIP if this is the compression applied to your files, or CSV if there is no compression. When choosing GZIP, your file must have a ".gz" or ".gzip" extension and contain only one file inside the archive. If you choose CSV, your file must have a ".csv" extension
5. Header: List the first line (header) of your files. You do not need to keep the column order. If a file has fields that are not in this field, those fields will be ignored. If the file does not have all the fields from the header, those fields will simply not be read, without causing failures. Do not use spaces between fields, only commas. For example: col_x,col_y,col_z

After this step, you will be able to choose the replication type for your integration.

If you choose Full, all files matching the prefix will always be read, and if a file is deleted between one run and another it will also be deleted from your table — which will not happen if your replication is Incremental, so this replication type is recommended for cases where files are deleted (not just changed). This replication can increase your record count

When choosing Incremental replication, whenever a file is changed it will be updated in your destination; you will be able to find which file generated the insertion of a given row and when that row was inserted/updated through the _kdd_file_name and _kdd_insert_time columns, respectively

S3 destination configuration and details

2. Output file format

We currently send files in the following formats:

2.1 JSON (jsonl)

This format saves data in JSON format, one on each line of the file, with quotes escaped by \. This is an example of a jsonl file that will be uploaded to your data lake:

{"\"keyword\"":"y","\"url\"":"https:\/\/mailchi.mp\/a37895aff5ac\/subscribe_newsletter","\"title\"":"https:\/\/mailchi.mp\/a37895aff5ac\/subscribe_newsletter","\"timestamp\"":[numero],"\"clicks\"":1}
{"\"keyword\"":"z","\"url\"":"https:\/\/pipedrivewebforms.com\/form\/43fd725e54683d565a44ed4872c3560f4639452","\"title\"":"Kondado - contato","\"timestamp\"":[numero],"\"clicks\"":2}

If you do not apply compression, the files will be saved with the extension “.json”

2.2 CSV

The CSV format will save files in this traditional format and with a “.csv” extension if you haven't provided any compression method. You will need to provide some additional file formatting parameters:

2.2.1 Column delimiter

Possible values for the column delimiter are

• Comma (",")
• Pipe("|")
• Tab("\t")

2.2.2 Quote

The quote parameter indicates how each column in a line of the file will be wrapped in quotes. Possible options are a 1 x 1 mapping with python's CSV module

• All
• minimal
• non-numeric
• None

We recommend the “All” format.

As you may have noticed, quote and the column delimiter can conflict with your content if these characters are also present in the values. So, all saved files have a default escape character for conflicting values, the backslash (“\”).

2.2.3 Header

You can also choose whether you want the uploaded files to come with a header or not. Remembering that, if you register a URL of notifications, sending a header becomes unnecessary, as we will send you the schema of the object, with the order of the columns according to how they were written in the file. Column order will also obey the same display order in our UI when you click on “DETAILS” of a given integration.

2.3 Parquet

Saves data in the parquet format. If compression is not used, the file extension will be .parquet

3. Compression

You can choose to apply a compression method to your files or not. The available compression methods will change the extension of saved files.

3.1 GZIP

Files compressed with the gzip method will receive the extension “.gz”

4. Notification URL

If the “Notify any URL” parameter is filled in, you will need to enter a valid URL in “URL for notifications”. This URL must be enabled to receive posts.

4.1 When will the URL notification occur?

At the end of an integration, for each object integrated and for each operation, we will make a post at the specified URL:

• “for each object” means that pipelines that populate more than one table will notify per table
• “for each operation” means that the send operations to the full tables and to the delta tables (if you chose to keep deltas) will make separate notifications

4.2 What if the URL fails?

We will consider as success a notification for the given URL that returns a 200 response code within 30 seconds. If the answer is different from this or takes longer, we will perform 3 more attempts and point out an error in the integration if the failure persists. However, files in full and deltas folders/prefixes will not be deleted

4.3 Payload Sent

With the data that will be sent to your notification URL, you will have all the necessary information to trigger further processes to handle the sent files. The default JSON payload fields are as follows:

4.3.1 pipeline_id

Id of the integration that generated the files. The id of an pipeline can be obtained through the URL of the Kondado pipeline in your browser: https://app.kondado.com.br/pipelines/pipeline_id

Example:

"pipeline_id":231923

4.3.2 bucket

Bucket where the files were uploaded. Corresponds to the bucket registered by you in the destination parameters at the time the integration saved the files

Example:

"bucket":"my-kdd-data-lake"

4.3.3 kdd_operation

Operation that triggered the notification. Possible values are:

• merge_deltas: indicates that this operation means sending data to the delta table
• merge_full: indicates that this operation means sending data to the full table

Example:

"kdd_operation":"merge_deltas"

4.3.4 kdd_replication_type

The replication type of this integration (full/incremental). For incremental merges, you should “upsert” the file values with your main file. For full integrations, new uploaded files must replace the previous core file.

The possible values of this parameter are:

• drop: indicates that the integration is of the integral type
• savepoint: indicates that the integration is of the incremental type

This parameter is sent only when kdd_operation=merge_full.

Example:

"kdd_replication_type":"savepoint"

4.3.5 kdd_replication_keys

If your integration is incremental (kdd_replication_type=savepoint), then this field will contain the fields (comma separated) that you should use to upsert with the main file.

Example:

"kdd_replication_keys":"lead_email,campaign_id"

4.3.6 schema

Array of jsons with the fields sent to the files, in the order they appear (useful for CSV and also for your main aggregator file to have a consistent typing with your data).

In this array, each json will correspond to a field in the destination file and will contain the following information:

db_name: name of the field in the destination. This is the value you should use to create your main file.

type: field typing, according to Kondado standardization. The field types that are currently supported by this target are:

• text
• timestamp
• gives you
• float
• int
• boolean
• team

key: field name according to Kondado connector/CMS. For the purposes of aggregating the files into a main file, this field can be ignored, we are sending it for clarification purposes only

Example:

"schema":[
       {
          "db_name":"lead_id",
          "type":"text",
          "key":"id"
       },
       {
          "db_name":"etapa",
          "type":"text",
          "key":"Etapas__PerguntasRespostas->Etapa"
       },
       {
          "db_name":"pergunta",
          "type":"text",
          "key":"Etapas__PerguntasRespostas->Pergunta"
       }
]

4.3.7 files

Array with available file names and their location (prefix). For example:

"files":[
      "page_hits_deltas/[numero].gz",
      "page_hits_deltas/[numero].gz"
   ]

4.3.8 file_format

Final file format created. Remember that, if the file undergoes compression, the format will not correspond to its extension, but the file “inside” the compressed file will be in this format. Example:

"file_format":"json"

4.3.9 file_format_params

If the selected file format requires additional formatting parameters (for example, CSV) this field will bring a JSON with the parameters used to create the file.

Example (CSV):

"file_format_params":{
    "csv_delimiter":"pipe",
    "csv_quote":"all",
    "csv_include_header":"yes"
}

4.3.10 file_compression

This parameter indicates the compression applied to the output file. If no compression has been applied, its value will be “no”.

Example:

"file_compression":"gzip"

4.3.11 kdd_row_count

This parameter indicates how many records/lines there are in total in the uploaded files.

Example:

"kdd_row_count":29173

5. Prefix

The name/final part of the file will always be the timestamp in which the integration was performed + extension, in this format: [YYYYMMDDhhmmssfffff] [.] [extension]

In addition to this fixed name, with this prefix parameter, you can add variables that will help organize your bucket, they are:

• [exec_year]: Year (YYYY) the integration was executed
• [exec_month]: Month (MM) of execution of the integration
• [exec_day]: Day of the month (DD) the integration was executed
• [exec_hour]: Time (hh 24) the integration was executed
• [exec_minute]: Minute (mm) of execution of the integration
• [exec_second]: Seconds (ss) of integration execution
• [exec_millisec]: Milliseconds (fffffff) of execution of integration

These variables can be combined with free text and will be concatenated in front of the file name during insertion into S3.

Attention: It is necessary to write the brackets – ] and [

Example

Consider that the following prefix has been filled in:

[exec_year]/month=[exec_month]/{day=[exec_day]}/

The final name of the file being this:

[numero].json

In this case, the name to be written in the destination table will be:

2021/month=09/{day=22}/[numero].json

Note that a slash “/” was used at the end of the prefix ([exec_year]/month=[exec_month]/{day=[exec_day]}/) – if this slash was not inserted, the file would be 2021/month=09/ {day=22}[numero].json which may not be what you want

6. Date Format

For date and timestamp type columns, you can specify the date format:

• ISO: values will be written in ISO 8601 format
• Epoch: Values will be written in unix epoch time format, in seconds

General inquiries

Tables or folders or prefixes?

When we send data to other databases, we call the table records “sets”. In the case of S3, the way to gather records that belong to the same category is by prefixes (commonly called “folders”). To maintain consistency with the rest of the targets we support, we'll continue to call these sets of records tables. This way, whenever you see “table” within Kondado referring to the S3 destination, understand it as “prefix” (or “folder”). So, when giving the name to a “full table” you will be saying what should be the prefix (folder) of the files created in “full” mode

If you wish, you can use “/” in the table names to create a subfolder structure.

The tables/folders are the “fixed” part of the files insertion, that is, the files will always be sent to the same bucket/folder path. The creation of “sub-folders” with the prefix parameter allows greater dynamism, being possible to specify different folders for years, months, days, times, etc., which will be changed according to the moment of execution of the pipeline.

File creation flow of an integration

When looking at your S3 bucket, you will come across several “folders” that have a similar nomenclature to this “kdd_xxxxxxx_staging”. Files with this prefix are currently being merged by Kondado. Once the source data reading is finished, these staging folders will contain all the files from this run. Kondado will then copy these files to the full and deltas tables – at the end of this copy, we will trigger the notification to the registered URL.

Before and after each integration, the staging tables will be emptied to receive new files. So, we ask you not to delete these staging tables, as their integration may fail or, worse, generate incomplete data.

On the Kondado platform, when a destination is a database, our platform handles all data de-duplication with each update and arrival of new data. However, in the case of S3, this flow is the responsibility of the user using the replication_keys fields. If you don't use the notification URL, these replication fields can be found in the connector's documentation, when accessing the relationship graph. They will be marked with a circle next to the field. In rare cases, it is possible for two similar files to be uploaded in the same run, so your deduplication flow should be able to handle this when checking new files for deduplication.