Pull and ingest data from a third-party API

This tutorial builds a data pipeline that pulls data from a third-party finance API and loads it into TimescaleDB.

This tutorial requires multiple libraries. This can make your deployment package size larger than the 250 MB limit of Lambda. You can use a Docker container to extend the package size up to 10 GB, giving you much more flexibility in libraries and dependencies. For more about AWS Lambda container support, see the AWS documentation.

The libraries used in this tutorial:

Create an ETL function

Extract, transform, and load (ETL) functions are used to pull data from one database and ingest the data into another. In this tutorial, the ETL function pulls data from a finance API called Alpha Vantage, and inserts the data into TimescaleDB. The connection is made using the values from environment variables.

This is the ETL function used in this tutorial:

  2. # function.py:
  4. import csv
  6. import pandas as pd
  8. import psycopg2
  10. from pgcopy import CopyManager
  12. import os
  14. config = {'DB_USER': os.environ['DB_USER'],
  16.          'DB_PASS': os.environ['DB_PASS'],
  18.          'DB_HOST': os.environ['DB_HOST'],
  20.          'DB_PORT': os.environ['DB_PORT'],
  22.          'DB_NAME': os.environ['DB_NAME'],
  24.          'APIKEY': os.environ['APIKEY']}
  26. conn = psycopg2.connect(database=config['DB_NAME'],
  28.                            host=config['DB_HOST'],
  30.                            user=config['DB_USER'],
  32.                            password=config['DB_PASS'],
  34.                            port=config['DB_PORT'])
  36. columns = ('time', 'price_open', 'price_close',
  38.           'price_low', 'price_high', 'trading_volume', 'symbol')
  40. def get_symbols():
  42.    """Read symbols from a csv file.
  44.    Returns:
  46.        [list of strings]: symbols
  48.    """
  50.    with open('symbols.csv') as f:
  52.        reader = csv.reader(f)
  54.        return [row[0] for row in reader]
  56. def fetch_stock_data(symbol, month):
  58.    """Fetches historical intraday data for one ticker symbol (1-min interval)
  60.    Args:
  62.        symbol (string): ticker symbol
  64.        month (int): month value as an integer 1-24 (for example month=4 fetches data from the last 4 months)
  66.    Returns:
  68.        list of tuples: intraday (candlestick) stock data
  70.    """
  72.    interval = '1min'
  74.    slice = 'year1month' + str(month) if month <= 12 else 'year2month1' + str(month)
  76.    apikey = config['APIKEY']
  78.    CSV_URL = 'https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY_EXTENDED&' \
  80.              'symbol={symbol}&interval={interval}&slice={slice}&apikey={apikey}' \
  82.              .format(symbol=symbol, slice=slice, interval=interval,apikey=apikey)
  84.    df = pd.read_csv(CSV_URL)
  86.    df['symbol'] = symbol
  88.    df['time'] = pd.to_datetime(df['time'], format='%Y-%m-%d %H:%M:%S')
  90.    df = df.rename(columns={'time': 'time',
  92.                            'open': 'price_open',
  94.                            'close': 'price_close',
  96.                            'high': 'price_high',
  98.                            'low': 'price_low',
  100.                            'volume': 'trading_volume'}
  102.                            )
  104.    return [row for row in df.itertuples(index=False, name=None)]
  106. def handler(event, context):
  108.    symbols = get_symbols()
  110.    for symbol in symbols:
  112.        print("Fetching data for: ", symbol)
  114.        for month in range(1, 2):
  116.            stock_data = fetch_stock_data(symbol, month)
  118.            print('Inserting data...')
  120.            mgr = CopyManager(conn, 'stocks_intraday', columns)
  122.            mgr.copy(stock_data)
  124.            conn.commit()

Add a requirements file

When you have created the ETL function, you need to include the libraries you want to install. You can do this by creating a text file in your project called requirements.txt that lists the libraries. This is the requirements.txt file used in this tutorial:

  2. pandas
  4. requests
  6. psycopg2-binary
  8. pgcopy
note

This example uses psycopg2-binary instead of psycopg2 in the requirements.txt file. The binary version of the library contains all its dependencies, so that you don’t need to install them separately.

Create the Dockerfile

When you have the requirements set up, you can create the Dockerfile for the project.

Creating the Dockerfile

  1. Use an AWS Lambda base image:

    1. FROM public.ecr.aws/lambda/python:3.8

  2. Copy all project files to the root directory:

    1. COPY function.py .
    3. COPY requirements.txt .

  3. Install the libraries using the requirements file:

    1. RUN pip install -r requirements.txt
    3. CMD ["function.handler"]

Upload the image to ECR

To connect the container image to a Lambda function, you need to upload it to the AWS Elastic Container Registry (ECR).

Uploading the image to ECR

  1. Log in to the Docker command line interface:

    1. aws ecr get-login-password --region us-east-1 \
    3. | docker login --username AWS \
    5. --password-stdin <AWS_ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com

  2. Build the image:

    1. docker build -t lambda-image .

  3. Create a repository in ECR. In this example, the repository is called lambda-image:

    1. aws ecr create-repository --repository-name lambda-image

  4. Tag your image using the same name as the repository:

    1. docker tag lambda-image:latest <AWS_ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/lambda-image:latest

  5. Deploy the image to Amazon ECR with Docker:

    1. docker push <AWS_ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/lambda-image:latest

Create a Lambda function from the container

To create a Lambda function from your container, you can use the Lambda create-function command. You need to define the --package-type parameter as image, and add the ECR Image URI using the --code flag:

  2. aws lambda create-function --region us-east-1 \
  4. --function-name docker_function --package-type Image \
  6. --code ImageUri=<ECR Image URI> --role <ARN_LAMBDA_ROLE>

Schedule the Lambda function

If you want to run your Lambda function according to a schedule, you can set up an EventBridge trigger. This creates a rule using a cron expression.

Scheduling the Lambda function

  1. Create the schedule. In this example, the function runs every day at 9 AM:

    1. aws events put-rule --name schedule-lambda --schedule-expression 'cron(0 9 * * ? *)'

  2. Grant the necessary permissions for the Lambda function:

    1. aws lambda add-permission --function-name <FUNCTION_NAME> \
    3. --statement-id my-scheduled-event --action 'lambda:InvokeFunction' \
    5. --principal events.amazonaws.com

  3. Add the function to the EventBridge rule, by creating a targets.json file containing a memorable, unique string, and the ARN of the Lambda Function:

    1. [
    3.   {
    5.     "Id": "docker_lambda_trigger",
    7.     "Arn": "<ARN_LAMBDA_FUNCTION>"
    9.   }
    11. ]

  4. Add the Lambda function, referred to in this command as the target, to the rule:

    1. aws events put-targets --rule schedule-lambda --targets file://targets.json
important

If you get an error saying Parameter ScheduleExpression is not valid, you might have made a mistake in the cron expression. Check the cron expression examples documentation.

You can check if the rule is connected correctly to the Lambda function in the AWS console. Navigate to Amazon EventBridge → Events → Rules, and click the rule you created. The Lambda function’s name is listed under Target(s):

Lamdba function target in AWS Console