Module 07: Connections & Extractors

Data reaches the platform from web sites, file stores, mailboxes and message streams. In ODSL there is no separate connection object to configure: each source is read with an active variable service, given the location as its id and any credentials as options. In this module you will read from the main kinds of source, keep the credentials in secrets, and see where extractors fit.

Learning Objectives​

By the end of this module you will be able to:

• Choose the right service for a source: HTTP, Azure Blob, Azure Files, FTP, S3, mail or Kafka
• Read files over HTTP(S) in CSV, JSON, XML, HTML and other formats
• Read from Azure Blob Storage with an account key kept in a secret
• Keep a copy of each source file with the archive service
• Explain what an extractor is and how it is managed
• Turn a working read into a scheduled process

Step 1: Choose a Service​

See External Data Collection Services for the full list.

Step 2: Read Over HTTP(S)​

The id is the URL. Options follow as strings:

//#region Read files over HTTPS
// XML: ECB reference rates
fx = ${xml:"https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml"}

// CSV with no header row, as a table
balloons = ${csv:"https://archive.ics.uci.edu/ml/machine-learning-databases/balloons/adult+stretch.data","headerpos=0","output=table"}
print balloons.rows.size

// A site that needs the curl tool
daily = ${curl:"https://www.example.com/prices/daily.csv"}
//#endregion

Useful CSV options are separator, headerpos, datapos and output. The XML service takes selector, ignoreContentType and xslt.

If a source needs an API token, read it from a secret:

//#region HTTP with a token
url = "https://web-api.tp.entsoe.eu/api?securityToken=" + ${secret:"ENTSOE_TOKEN"}
//#endregion

Step 3: Read from Azure Blob Storage​

The id is the blob URL (account, container and path). The account key goes in the AccountKey option, taken from a secret:

//#region Read a blob
key = ${secret:"nge-blob-key"}
prices = ${azblob:"https://ngestorage.blob.core.windows.net/prices/daily/ttf.csv","AccountKey=" + key.value}
print prices
//#endregion

The file is converted with the service that matches its type, here CSV. The service can also write: saving a variable with the blob URL in the url option writes it as JSON.

Step 4: Read from a Mailbox or Kafka​

Many suppliers still email a daily file. The mail service reads the attachments of matching emails:

//#region Read emailed attachments
pwd = ${secret:"PRICES_MAILBOX_PASSWORD"}
files = ${mail:"imap.example.com","user=prices@company.com,password=" + pwd.value + ",sender=settlements@example-broker.com,subject=Daily settlement prices"}
print files.size
//#endregion

Options include agent (imap or pop3), port, folder, seen, sender, subject, fromDateTime, toDateTime and filter_files.

The Kafka service returns the messages published since the consumer group last read the topic:

//#region Read a Kafka topic
messages = ${kafka:"prices-*","bootstrap=b-1.example.com:9196;b-2.example.com:9196,group=nge_prices,username=nge_reader"}
//#endregion

Without a password option, Kafka uses the KAFKA_SASL_PASSWORD environment variable.

Step 5: Keep a Copy of the Source​

The archive service downloads a file over HTTP, HTTPS or SFTP and stores it in your file archive, so you can see later exactly what a loader read:

//#region Archive the source file
status = ${archive:"https://www.example.com/prices/daily.csv"}
print status
//#endregion

Step 6: Extractors​

Some sites need navigation (filling forms, clicking buttons) before the data can be read. An extractor is a pipeline of instructions sent to the website to do this. Extractors are stored in the extractor service and are versioned automatically, like scripts and transformers.

Extractors belong to an environment, so build and test them in test and promote them with the Promotion extension. See Extractor.

Step 7: Schedule the Feed​

Once a script reads, transforms and loads correctly, run it as a process on a schedule:

//#region Schedule a loader
p = Process()
p.service = "ETL"
p.name = "SUPPLIER_DAILY_PRICES"
p.description = "Load the supplier daily price file"
p.scriptName = 'loaders\supplier_daily_prices'
p.cron = "0 19 ? * MON-FRI *"
p.enabled = true
save ${process:p}
//#endregion

Step 8: MongoDB Connections​

If your company uses its own MongoDB Atlas cluster for private data, or named connections to other clusters, these are part of the tenant configuration: the mongodb and connections parts of the tenant, which must start with mongodb. See Database Information.

Exercise​

1. Read the ECB XML feed and print the first currency and rate.
2. Create a secret with a placeholder storage key and write the azblob read that would use it.
3. Archive the ECB file with the archive service and check the returned status.
4. List the extractors in your environment and the versions of one of them.

Summary​