• S3cmd Setup
• S3 API
• Overview
• Design Considerations
• Create Hive Table Catalog
• Supported File Types
• Password Options
• S3 GIMEL Read API for CSV
• S3 GIMEL Write API CSV
• S3 GIMEL GSQL
• Limitations

---

S3cmd Setup

• This CLI is used to access S3 from command line.
• Download s3cmd binary from https://s3tools.org/s3cmd.
• Configure it by creating .s3cfg file in your home folder.
  Edit Access ID, Secret Key and S3 End point in the configuration file.

[default] access_key = [ACCESS-ID] access_token = add_encoding_exts = add_headers = bucket_location = region ca_certs_file = cache_file = check_ssl_certificate = True check_ssl_hostname = True cloudfront_host = cloudfront.amazonaws.com content_disposition = content_type = default_mime_type = binary/octet-stream delay_updates = False delete_after = False delete_after_fetch = False delete_removed = False dry_run = False enable_multipart = True encrypt = False expiry_date = expiry_days = expiry_prefix = follow_symlinks = False force = False get_continue = False gpg_command = None gpg_decrypt = %(gpg_command)s -d --verbose --no-use-agent --batch --yes --passphrase-fd %(passphrase_fd)s -o %(output_file)s %(input_file)s gpg_encrypt = %(gpg_command)s -c --verbose --no-use-agent --batch --yes --passphrase-fd %(passphrase_fd)s -o %(output_file)s %(input_file)s gpg_passphrase = noencryption guess_mime_type = True host_base = [S3 End Point] host_bucket = %(bucket).s3-region.example.com human_readable_sizes = False invalidate_default_index_on_cf = False invalidate_default_index_root_on_cf = True invalidate_on_cf = False kms_key = limit = -1 limitrate = 0 list_md5 = False log_target_prefix = long_listing = False max_delete = -1 mime_type = multipart_chunk_size_mb = 10 multipart_max_chunks = 10000 preserve_attrs = True progress_meter = True proxy_host = proxy_port = 0 put_continue = False recursive = False recv_chunk = 65536 reduced_redundancy = False requester_pays = False restore_days = 1 restore_priority = Standard secret_key = [SECRET-KEY] send_chunk = 65536 server_side_encryption = False signature_v2 = False signurl_use_https = False simpledb_host = sdb.amazonaws.com skip_existing = False socket_timeout = 300 stats = False stop_on_error = False storage_class = throttle_max = 100 upload_id = urlencoding_mode = normal use_http_expect = False use_https = False use_mime_magic = True verbosity = WARNING website_endpoint = http://%(bucket)s.s3-website-%(location)s.amazonaws.com/ website_error = website_index = index.html

• Test the connection

./s3cmd info s3://gimeltestbucket

• List the file on S3 bucket

./s3cmd ls --recursive s3://gimeltestbucket

• Put file on S3 bucket

./s3cmd put test s3://gimeltestbucket/test

• Download file from S3 bucket

./s3cmd get s3://gimeltestbucket/test

---

S3 API

Overview

• This API will enable read, write objects into/from S3 storage

---

Design Considerations

Spark S3 connector

• https://docs.databricks.com/spark/latest/data-sources/aws/amazon-s3.html.
• Spark S3 uses aws-java-sdk and hadoop-aws libraries for constructing dataframes by downloading files from S3 buckets and writing dataframe to S3 buckets.
• Gimel connector is using these libraries as dependencies.

---

Create Hive Table Catalog

Create Hive DDLs when using HIVE as gimel.catalog.provider.
The following hive table points to S3 bucket - gimeltestbucket for reading/writing files to/from an object location

CSV file

  CREATE EXTERNAL TABLE `udc.test_s3_airports_csv`(
    `data` string COMMENT 'from deserializer')
  LOCATION
    'hdfs://hadoopcluster/tmp/udc/test_s3_airports_csv'
  TBLPROPERTIES (
    'gimel.s3.object.format'='csv',
    'gimel.s3.object.location'='s3a://gimeltestbucket/flights/airports.csv',
    'gimel.s3.file.header'='true',
    'gimel.s3.file.inferSchema'='true',
    'gimel.storage.type'='S3');

Parquet file

  CREATE EXTERNAL TABLE `udc.test_s3_parquet`(
    `data` string COMMENT 'from deserializer')
  LOCATION
    'hdfs://hadoopcluster/tmp/udc/test_s3_parquet'
  TBLPROPERTIES (
    'gimel.s3.object.format'='parquet',
    'gimel.s3.object.location'='s3a://gimeltestbucket/userdata.parquet',
    'gimel.storage.type'='S3');

Json file

  CREATE EXTERNAL TABLE `udc.test_s3_json`(
    `data` string COMMENT 'from deserializer')
  LOCATION
    'hdfs://hadoopcluster/tmp/udc/test_s3_json'
  TBLPROPERTIES (
    'gimel.s3.object.format'='json',
    'gimel.s3.object.location'='s3a://gimeltestbucket/test.json',
    'gimel.storage.type'='S3');

---

Create JSON Dataset Properties

Create DatasetProperties json when using USER as gimel.catalog.provider.

val dataSetProperties_s3 = s"""
{
      "datasetType": "S3",
      "fields": [],
      "partitionFields": [],
      "props": {
            "gimel.storage.type":"S3",
            "datasetName":"udc.sample_s3_dataset",
            "fs.s3a.path.style.access":"true",
            "fs.s3a.endpoint":"s3-region.example.com",
            "gimel.s3.credentials.strategy": "file",
            "gimel.s3.credentials.file.path": "/path/s3_credentials",
            "gimel.s3.credentials.file.source": "local",
            "gimel.s3.file.header": "true",
            "gimel.s3.file.inferSchema": "true"

         }
 }"""

---

Supported File Types

• CSV
• JSON
• PARQUET
• TXT

---

Catalog Properties

Property	Mandatory?	Description	Example	Default
fs.s3a.path.style.access	N	Path style enabled in S3?	true/false	true
fs.s3a.endpoint	Y	S3 end point	s3-us.example.com:80
fs.s3a.access.key	Y	Access ID (This is a Must if gimel.s3.credentials.strategy=user)
fs.s3a.secret.key	Y	Secret Key (This is a Must if gimel.s3.credentials.strategy=user)
gimel.s3.file.header	N	File Header for CSV file	true/false	false
gimel.s3.file.inferSchema	N	Infer Schema from header?	true/false	false
gimel.s3.object.location	Y	Object Location on S3	s3a://gimeltestbucket/flights/airports.csv
gimel.s3.credentials.strategy	Y	Credentials strategy	file/user/credentialLess	file
gimel.s3.credentials.file.source	Y	This is a Must if gimel.s3.credentials.strategy=file	local/hdfs
gimel.s3.credentials.file.path	Y	This is a Must if gimel.s3.credentials.strategy=file	/path/xxxx/s3_credentials
gimel.s3.save.mode	N	Write mode for S3 (https://spark.apache.org/docs/2.3.0/sql-programming-guide.html#save-modes)	append/overwrite/error/ignore	error
--------------------------------------------------------------------------------------------------------------------

Password Options

• To access an S3 bucket, Access ID and Secret Key are required
• These credentials can be given using either local file sytem or HDFS file system
• For local file system, we need to put the credentials in a file and mention the file as below options

• This will be useful in default or yarn client mode

  "gimel.s3.credentials.strategy" -> "file" "gimel.s3.credentials.file.source" -> "local" "gimel.s3.credentials.file.path" -> "/path/xxxx/mycredentials.txt"

• for hdfs file system, we need to put the password in a file and mention the file as below options.

• This will be useful in yarn cluster mode

  "gimel.s3.credentials.strategy" -> "file" "gimel.s3.credentials.file.source" -> "hdfs" "gimel.s3.credentials.file.path" -> "hdfs://cluster/xxxx/mycredentials.txt"

• The credentials file must not be accessible by any other user (Set permission to 700)

• Put the Access ID and Secret Key in credentials file in the following format ACCESS-ID SECRET-KEY

S3 GIMEL Read API for CSV

• The following example says how to read data from S3 by giving the credentials using local file system as source

Common Imports and initializations

val dataSet = new com.paypal.gimel.DataSet(spark)

val options = Map("gimel.s3.object.format" -> "csv", 
                   "gimel.s3.credentials.strategy" -> "file",
                   "gimel.s3.credentials.file.source" -> "local",
                   "gimel.s3.credentials.file.path" -> "/path/xxxx/s3_credentials",
                   "gimel.s3.file.header" -> "true",
                   "gimel.s3.file.inferSchema" -> "true")

val dfRead = dataSet.read("udc.S3.Test.gimeltestbucket.flights_airports_csv", options)
dfRead.show

---

S3 GIMEL Write API CSV

• The following example says how to write data to S3 by giving the credentials using HDFS file as source

val options = Map("gimel.s3.object.format" -> "csv", 
                   "gimel.s3.credentials.strategy" -> "file",
                   "gimel.s3.credentials.file.source" -> "hdfs",
                   "gimel.s3.credentials.file.path" -> "hdfs://hadoopcluster/user/xxxx/s3_credentials",
                   "gimel.s3.file.header" -> "true",
                   "gimel.s3.file.inferSchema" -> "true")

val dfWrite = dataSet.write("udc.S3.Test.gimeltestbucket.test_kafka_to_s3", options)

S3 GIMEL GSQL

Common Imports and initializations

import org.apache.spark.sql.{Column, Row, SparkSession,DataFrame}
import org.apache.spark.sql.functions._

// Create Gimel SQL reference
val gsql: (String) => DataFrame = com.paypal.gimel.sql.GimelQueryProcessor.executeBatch(_: String, spark)

//Set UDC parameters for testing
gsql("set rest.service.method=https")
gsql("set rest.service.host=udc-rest-api-host")
gsql("set rest.service.port=443")

gsql("set gimel.catalog.provider=UDC")
gsql("set gimel.logging.level=CONSOLE")

CSV file Read

gsql("set gimel.s3.object.format=csv")
gsql("set gimel.s3.credentials.strategy=file")
gsql("set gimel.s3.credentials.file.source=local")
gsql("set gimel.s3.credentials.file.path=/path/xxxxx/s3_credentials")
gsql("set gimel.s3.file.header=true")
gsql("set gimel.s3.file.inferSchema=true")

val df = gsql("select * from udc.S3.Test.gimeltestbucket.flights_airports_csv")

JSON file Read

gsql("set gimel.s3.object.format=json")
gsql("set gimel.s3.credentials.strategy=file")
gsql("set gimel.s3.credentials.file.source=local")
gsql("set gimel.s3.credentials.file.path=/path/xxxx/s3_credentials")

val df = gsql("select * from udc.S3.Test.gimeltestbucket.test_json")

CSV file Read with different delimeter

gsql("set gimel.s3.object.format=csv")
gsql("set gimel.s3.file.delimiter=|")
gsql("set gimel.s3.credentials.strategy=file")
gsql("set gimel.s3.credentials.file.source=local")
gsql("set gimel.s3.credentials.file.path=/path/xxxxx/s3_credentials")
gsql("set gimel.s3.file.header=true")
gsql("set gimel.s3.file.inferSchema=true")

val df = gsql("select * from udc.S3.Test.gimeltestbucket.test_delimiter_csv")

Parquet file Read

gsql("set gimel.s3.object.format=parquet")
gsql("set gimel.s3.credentials.strategy=file")
gsql("set gimel.s3.credentials.file.source=local")
gsql("set gimel.s3.credentials.file.path=/path/xxxx/s3_credentials")

val df = gsql("select * from udc.S3.Test.gimeltestbucket.userdata_parquet")

Kafka to S3

gsql("set gimel.s3.object.format=csv")
gsql("set gimel.s3.credentials.strategy=file")
gsql("set gimel.s3.credentials.file.source=local")
gsql("set gimel.s3.credentials.file.path=/path/xxxxx/s3_credentials")
gsql("set gimel.s3.file.header=true")
gsql("set gimel.s3.file.inferSchema=true")
gsql("set gimel.kafka.throttle.batch.fetchRowsOnFirstRun=1000")
gsql("set gimel.s3.save.mode=overwrite")

val df = gsql("insert into udc.S3.Test.gimeltestbucket.flights_airports_csv 
                select * from udc.Kafka.Gimel_Dev.test.flights_airports")

Limitations

• Presently we dont support Binary file type.