Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Jason Myers - Leveraging Serverless Architectur...

Jason Myers - Leveraging Serverless Architecture for Powerful Data Pipelines

Serverless Architectures that allow us to run python functions in the cloud in an event-driven parallel fashion can be used to create extremely dynamic and powerful data pipelines for use in ETL and data science. Join me for an exploration of how to build data pipelines on Amazon Web Services Lambda with python. We'll cover a single introduction to event-driven programming. Then, we'll walk through building an example pipeline while discussing some of the frameworks and tools that can make building your pipeline easier. Finally, we'll discuss how to maintain observability on your pipeline to ensure proper performance and troubleshooting information.

https://us.pycon.org/2017/schedule/presentation/566/

PyCon 2017

May 21, 2017
Tweet

More Decks by PyCon 2017

Other Decks in Programming

Transcript

  1. Project Structure !"" functions # $"" listener # $"" main.py

    !"" infrastructure # !"" dev # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf # !"" prod # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf !"" project.json $"" project.prod.json
  2. Project Structure !"" functions # $"" listener # $"" main.py

    !"" infrastructure # !"" dev # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf # !"" prod # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf !"" project.json $"" project.prod.json
  3. Project Structure !"" functions # $"" listener # $"" main.py

    !"" infrastructure # !"" dev # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf # !"" prod # # !"" main.tf # # !"" outputs.tf # # $"" variables.tf !"" project.json $"" project.prod.json
  4. Apex package.json { "name": "listener", "description": "S3 File Listener", "runtime":

    "python3.6", "memory": 128, "timeout": 5, "role": "arn:aws:iam::ACOUNTNUM:role/listen_lambda_function", "environment": {}, "defaultEnvironment": "dev" }
  5. S3 Event Handler import logging import boto3 log = logging.getLogger()

    log.setLevel(logging.DEBUG) def get_bucket_key(event); bucket = event['Records'][0]['s3']['bucket']['name'] key = event['Records'][0]['s3']['object']['key'] return bucket, key def handle(event, context): log.info('{}-{}'.format(event, context)) bucket_name, key_name = get_bucket_key(event)
  6. S3 Event Handler import logging import boto3 log = logging.getLogger()

    log.setLevel(logging.DEBUG) def get_bucket_key(event); bucket = event['Records'][0]['s3']['bucket']['name'] key = event['Records'][0]['s3']['object']['key'] return bucket, key def handle(event, context): log.info('{}-{}'.format(event, context)) bucket_name, key_name = get_bucket_key(event)
  7. S3 Event Handler import logging import boto3 log = logging.getLogger()

    log.setLevel(logging.DEBUG) def get_bucket_key(event); bucket = event['Records'][0]['s3']['bucket']['name'] key = event['Records'][0]['s3']['object']['key'] return bucket, key def handle(event, context): log.info('{}-{}'.format(event, context)) bucket_name, key_name = get_bucket_key(event)
  8. S3 Event Handler (cont.) values = { 'bucket_name': bucket_name, 'key_name':

    key_name, 'timestamp': datetime.utcnow().isoformat() } client = boto3.client('sqs') client.publish( TopicArn=topic_arn, Message=json.dumps(values) )
  9. AWS Logging Permissions data "aws_iam_policy_document" "listener_logging" { statement { sid

    = "AllowRoleToOutputCloudWatchLogs" effect = "Allow" actions = ["logs:*"] resources = ["*"] } } resource "aws_iam_policy" "listener_logs" { name = "listener_logs" description = "Allow listener to log operations" policy = "${data.aws_iam_policy_document.listener_logging.json}" }
  10. AWS IAM Role Assumption data "aws_iam_policy_document" "listener_lambda_assume_role" { statement {

    sid = "AllowRoleToBeUsedbyLambda" effect = "Allow" actions = ["sts:AssumeRole"] principals { type = "Service" identifiers = ["lambda.amazonaws.com"] } } } resource "aws_iam_role" "listener_lambda_function" { name = "listener_lambda_function" assume_role_policy = "${data.aws_iam_policy_document.listener_lambda_assume_role.json}" }
  11. AWS Policy A!achment resource "aws_iam_policy_attachment" "listener_logs_attach" { name = "listener_logs_attach"

    roles = ["${aws_iam_role.listener_lambda_function.name}"] policy_arn = "${aws_iam_policy.listener_logs.arn}" }
  12. Lambda Packages List bcrypt cffi PyNaCl datrie LXML misaka MySQL-Python

    numpy OpenCV Pillow (PIL) psycopg2 PyCrypto cryptography pyproj python-ldap python- Levenshtein regex