DEV Community: Sam

Running Notebooks with arbitrary dependencies in an AWS lambda

Sam — Fri, 13 Oct 2023 05:18:18 +0000

Goals:

Run notebooks files in a lambda.
Allow them to install their own dependencies.

In this instance, I've used serverless framework, but the problems solved likely apply to other frameworks. After trying a number of approaches, the following seemed to work within the constraints of lambdas read-only file system:

Create a dedicated workspace in /tmp.
Copy the notebook and a script to start and execute a virtual environment into the workspace.
Fork off to the script and allow it to run to completion.

Starting with the serverless.yml file, note "IPYTHONDIR" must be set to somewhere in /tmp since lambdas run on a read only file system:

service: nb-exec
frameworkVersion: '3'
provider:
  name: aws

functions:
  hello:
    handler: handler.hello
    environment:
      IPYTHONDIR: /tmp/ipythondir

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    fileName: requirements.txt
    dockerizePip: true

package:
  patterns:
    - "!.venv/**"
    - "!node_modules/**"

Our requirements.txt file, which we will use to execute the notebook files:

nbconvert===7.9.2
ipython===8.16.1
ipykernel===6.25.2

Next, inside our handler:

import os
import shutil
import subprocess
import uuid


def hello(event, context):
    unique_id = str(uuid.uuid4())

    workspace_path = os.path.join(os.path.abspath(os.sep), "tmp", f"workspace_{unique_id}")
    if not os.path.exists(workspace_path):
        os.makedirs(workspace_path)

    shutil.copy("execute.sh", workspace_path)

    notebook_dir_path = os.path.join(workspace_path, "notebook")
    os.makedirs(notebook_dir_path, exist_ok=True)

    shutil.copy("example.ipynb", notebook_dir_path)

    execute_script_path = os.path.join(workspace_path, "execute.sh")
    subprocess.run(["bash", execute_script_path], cwd=workspace_path)

And finally the execute.sh file:

# Make sure dependencies can be picked up from the deployment directory, as well as the
# built in AWS runtime dependencies.
export PYTHONPATH=$LAMBDA_TASK_ROOT:$LAMBDA_RUNTIME_DIR

# Create a virtual environment that inherits these dependencies.
python3 -m venv .venv --system-site-packages
source .venv/bin/activate

python3 -m nbconvert --to notebook --execute ./notebook/example.ipynb

One unsolved additional problem is the following error when installing dependencies from within a cell:

!pip install pandas
Error: out of pty devices

But replacing this with the following seems to work fine:

subprocess.run(["pip", "install", "pandas"])

Note, running untrusted code in a lambda environment is not secure as each invocation may have access to other invocations or AWS resources.

Packaging and deploying Zappa applications on M1 silicon

Sam — Fri, 06 Oct 2023 04:56:17 +0000

When deploying Zappa applications, it's possible to package a version during deployment that can manifest is a bunch of missing dependency errors:

[ERROR] ImportError: Unable to import required dependencies:
numpy: Error importing numpy: you should not try to import numpy from
        its source directory; please exit the numpy source tree, and relaunch
        your python interpreter from there.
[ERROR] ImportError: Unable to import required dependencies:
numpy: 

IMPORTANT: PLEASE READ THIS FOR ADVICE ON HOW TO SOLVE THIS ISSUE!

Importing the numpy C-extensions failed. This error can happen for
many reasons, often due to issues with your setup or how NumPy was
installed.
[ERROR] ImportError: No module named 'sklearn.__check_build._check_build'
It seems that scikit-learn has not been built correctly.

Zappa claims to:

replace any dependencies with versions with wheels compatible with lambda

...but this doesn't seem to be the case (at least when building on apple M1 silicon). A fix I've found is to package and deploy within a docker container, which seems to work:

# Shell into a container.
docker run --platform linux/amd64 -v $(pwd):/builds/ -it python:3.9.18-slim-bullseye /bin/bash

# Create a virtual env to install dependencies.
cd /builds/
python3 -m venv .venv && source .venv/bin/activate

# Install the dependencies and zappa.
pip install -r requirements.txt
pip install zappa

# Add credentials and deploy.
mkdir ~/.aws && echo "creds" > ~/.aws/credentials
zappa deploy dev

Ensure that the environment within the container matches that of your lambda deployment:

Confirm the 'Architecture' is the same by running arch within the container and comparing to the above.
Ensure the python version matches by running python --version and selecting the correct docker container from this list: https://hub.docker.com/_/python/tags

SNS Email notifications for errors trigged by AWS Lambda functions using CloudWatch Alarms and CDK

Sam — Fri, 16 Jun 2023 06:01:07 +0000

The following is a CDK stack, setup for monitoring errors logged from Lambda functions, to supplement the lack of good examples for achieving this.

It is setup with the following goals but can be customised:

Measure at 1 minute intervals.
Trigger alerts if any errors are logged.
Send to an email address.

import * as cdk from "@aws-cdk/core";
import {Duration} from "@aws-cdk/core";
import * as sns from "@aws-cdk/aws-sns";
import * as subs from "@aws-cdk/aws-sns-subscriptions";
import * as cw from "@aws-cdk/aws-cloudwatch";
import {ComparisonOperator} from "@aws-cdk/aws-cloudwatch";
import * as actions from "@aws-cdk/aws-cloudwatch-actions";

export class CloudwatchAlarmsStack extends cdk.Stack {
  constructor(scope: cdk.Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const topic = this.createTopic('foo@example.com');
    this.alertOnLambdaProducesAnyError(topic, 'lambda-a');
    this.alertOnLambdaProducesAnyError(topic, 'lambda-b');
  }

  createTopic(notifyEmail: string) {
    const topic = new sns.Topic(this, 'oly-monitoring', {
      displayName: 'OlyServiceMonitoring',
    });
    topic.addSubscription(
        new subs.EmailSubscription(notifyEmail),
    );
    return topic;
  }

  alertOnLambdaProducesAnyError(topic: sns.Topic, lambdaName: string) {
    const alarm = new cw.Alarm(this, `any-errors-${lambdaName}`, {
      alarmName: `any-errors-${lambdaName}`,
      actionsEnabled: true,
      alarmDescription: `Notify for any errors logged for the lambda function ${lambdaName}`,
      threshold: 1,
      evaluationPeriods: 1,
      comparisonOperator: ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,

      metric: new cw.Metric({
        namespace: 'AWS/Lambda',
        metricName: 'Errors',
        dimensionsMap: {
          'FunctionName': lambdaName,
        },
        statistic: 'min',
        period: Duration.minutes(1),
      }),
    });

    alarm.addAlarmAction(new actions.SnsAction(topic));
  }

}

Event Sourcing: a one year retrospective

Sam — Fri, 28 Oct 2022 01:46:37 +0000

It has been a little over a year since building and supporting a production service based on Event Sourcing, where instead of designing a normalised relational database schema, the persistence layer of our domain was modelled as an immutable event stream.

With this approach, the read and write models are independent, you can create any pattern of access that is useful to your application, be it a relational database schema or something more specialised.

Our simplest example is the "wallet" aggregate, which records events based on receiving and spending credits. In the following example, we see the wallet event stream and the database table the events are projected into, noting that the event stream has far more information and context than our read model actually cares about (the total number of credits the user has at any one time).

+------------------------+--------+----------------+---------------------------------------------------------+
| Time                   | Stream | Event          | Payload                                                 |
+------------------------+--------+----------------+---------------------------------------------------------+
| 9/02/2022 02:44:19 PM  | wallet | CreditsGranted | {                                                       |
|                        |        |                |     "amount": 1                                         |
|                        |        |                | }                                                       |
| 10/02/2022 08:38:48 AM | wallet | CreditsSpent   | {                                                       |
|                        |        |                |     "amount": 1,                                        |
|                        |        |                |     "type": "jobLaunch",                                |
|                        |        |                |     "typeMetadata": {                                   |
|                        |        |                |         "jobId": "9348e9b3-43b8-4531-b79c-d70d0cb66ba0" |
|                        |        |                |     }                                                   |
|                        |        |                | }                                                       |
+------------------------+--------+----------------+---------------------------------------------------------+

MariaDB [main]> select aggregate_id, credits from account_projection where credits > 0;
+--------------------------------------+---------+
| aggregate_id                         | credits |
+--------------------------------------+---------+
| 911357c8-4b63-4e5a-8b5d-d38eff30a7cb |       7 |
| ff452810-32d3-483b-9d93-064a1df11190 |       9 |
+--------------------------------------+---------+

Our Wallet aggregate, responsible for recording events CreditsGranted and CreditsSpent:

final class Wallet implements AggregateRoot {
    use AggregateRootBehaviourWithRequiredHistory;

    private int $totalCredits = 0;

    public static function create(WalletId $id): static {
        return new static($id);
    }

    public function grantCredits(int $amount): void {
        if ($amount < 1) {
            throw CreditAddException::because('Cannot grant an amount of credits less than 1');
        }
        $this->recordThat(new CreditsGranted($amount));
    }

    protected function applyCreditsGranted(CreditsGranted $event): void {
        $this->totalCredits += $event->amount;
    }

    public function getTotalCredits(): int {
        return $this->totalCredits;
    }

    public function spendCredits(int $amount, CreditSpendTypeInterface $spendType): void {
        if ($amount < 1) {
            throw CreditSpendException::because('Cannot spend an amount of credits less than 1');
        }
        if ($amount > $this->totalCredits) {
            throw CreditSpendException::because('Not enough credits in wallet');
        }
        $this->recordThat(new CreditsSpent($amount, $spendType));
    }

    protected function applyCreditsSpent(CreditsSpent $event): void {
        $this->totalCredits -= $event->amount;
    }
}

With quick tour out of the way, what are some of the pros and cons of this pattern, having used it for a year.

Flexible read models 😁

Being able to quickly design a schema and rewrite it later has been extremely useful. When rapidly prototyping features, like the wallet above, our first version only validates the users credit balance when attempting to launch a job on our platform, although at any time we could expand the read model to a history of the transactions.

Another tangible benefit has been doing away with the notion of DRY database design. Our application is a two sided marketplace, where parties transact around a "job" aggregate. The access each party gets to various information related to the job during its lifecycle is vastly different. Instead of having a single entity that is carefully access controlled and guarded we project into two completely different "job owner" and "job participant" schemas, which vastly simplifies the access control and complexity of our application. When working on the experience for either party, you have a read model tailored to each persona, that you can confidently pass to them for consumption.

Audit trails 😁

A real struggle with traditional database schemas is figuring out how much information is worth keeping at a given moment in time. Do you want to store when something was created? Probably. What about when it was last updated? Yeah, that seems useful to. What about when a single property changed? Maybe, maybe not? What about timestamps for each change of each property along with who initiated the change? Certainly not.

The event stream brings time into the equation as a first class citizen. Instead of having timestamp columns or logs that provide a narrow insights into what's going on, a collection of unchanging events show you when an event occurred and who initiated it, in all cases.

This is helpful for auditing, but also I've noticed our app seems to more and more frequently bubble up information about who did something and when, which I think has measurable improved some of our interfaces. An example from the front-end of our ACL system:

More tooling 😐

With a read and write model, plus tools to orchestrate rebuilding the write models, there is certainly more tooling and code required to implement this pattern.

There will also be a stage where there is more investment required. At the moment, we truncate and rebuild all our projections, every deploy, since at our current volume of events it takes < 30 seconds. At some future point in time we may need to be more selective, rebuild them more iteratively or consider additional build steps to streamline rebuilding into a new schema.

Excerpts of our rebuild command:

    bin/console dpa:rebuild-projections
    Rebuilding projections
    +----------------------+---------------+
    | Aggregate            | Message Count |
    +----------------------+---------------+
    | access_control_list  | 1880          |
...
    | wallet               | 16            |
    | total                | 7758          |
    +----------------------+---------------+

    Finished batch 0 default/batch0: 32.00 MiB - 917 ms
...
    Finished batch 14 default/batch14: 40.00 MiB - 958 ms

    Total events: 7,758 in default/rebuild: 42.00 MiB - 26310 ms
    Projections rebuilt
    bin/console dpa:update-projection-status --done
    Existing status: rebuilding
    New status: up to date

Poor choice for some domains 😐

There are some domains which I think the pattern hasn't added much value. So far, I think that any time we're attempting to pull, sync or represent information from another service, system or process orchestrated elsewhere, using Event Sourcing adds an overhead that doesn't provide much benefit.

The source of truth is already housed somewhere else and understanding when a piece of information crossed the boundary from one service into another is less useful.

Thankfully pivoting away from Event Sourcing in such cases is extremely easy: project into a schema, swap your application code to read and write directly to the new schema, cease rebuilding the projection.

With these caveats in mind, I do still I think the pattern has been extremely useful and has paid back any investment many times over.

If you are interested in implementing this pattern, I found EventSauce to be a fantastic reference implementation, community and learning resource.

Header generated by DALL-E with the prompt: A wallet with digital features, with envelopes orderly stacked within it, digital art

Trialing Cloudflare R2 for zero cost egress

Sam — Wed, 05 Oct 2022 08:44:52 +0000

At Humyn.ai, our challenges sometimes involve distributing large datasets to our innovator community. For a recent challenge, we estimated ~$25 in charges for data being downloaded from S3, per participant, prompting investigation into other options.

Recently Cloudflare made their object storage product R2 generally available, advertising zero cost egress. The positioning of the product, taking direct aim at the relatively expensive cost of outgoing bandwidth from the other major cloud providers.

Our experience trialing R2 for disseminating large datasets follows.

Cost savings

In this case, the cost savings were potentially significant, but R2 does not provide any metrics on how much data was downloaded from a given bucket, however for anyone with a history of billing for a comparable product like S3, it should be easy to evaluate the impact of switching on the bottom line.

Supported clients

R2 (and apparently a number of other providers) implement the S3 API, allowing S3 clients to interact with R2 directly. I found that while the aws-cli worked for smaller files, there was an incompatibility with larger files using multi-part uploads.

I found a post from a Cloudflare engineer, that confirmed it was a known issue, but the configuration options in the aws cli to reduce concurrency did not seem to work.

In the end I switched to rclone with a limited concurrency of 1, which worked for files in the ~250GB range.

Features

Both pre-signed links and public buckets were supported in R2, which were enough for our short term needs.

S3 has extremely flexible IAM, amongst a wide range of other features and it was a little surprising that the tokens required to authenticate with R2 were global account level tokens, which couldn't be scoped per bucket.

With R2 being such a new product, I wouldn't be surprised if it continued to chip away at the advanced use cases supported by S3.

Overall our experience with R2 was positive and we'll be keeping an open mind on how it can be leveraged in our stack moving forward.

Decoding and validating AWS Cognito JWTs with PHP

Sam — Fri, 08 Apr 2022 03:13:06 +0000

Skip to the working sample repository or read on for explanation...

At Unearthed, we use the AWS service Cognito to issue JWTs to clients during authentication. From there, the JWT is exchanged with whichever services the user is interacting with in order to validate their identity. This is helpful when building out a service graph, since each JWT can describe an authenticated users session without any direct dependency on a user service.

If you are decoding Cognito JWTs from PHP, you'll need a few key pieces of information to start:

The region your Cognito user pool is based in (us-east-2 in our case).
The User Pool ID the token was issued from.
The Client ID used to issue the token.

To decode the tokens we'll be using web-token/jwt-framework and a few other dependencies that can be installed with:

composer require web-token/jwt-checker web-token/jwt-signature-algorithm-rsa guzzlehttp/guzzle

The process of decoding and validating a token is:

Download the public keys used to sign the token.
Decode the token and validate it against the public keys.
Verify the claims in the token.

Downloading the keys

Our Cognito configuration can be represented as a simple value object:

<?php

namespace Sam\JwtBlogPost;

class CognitoConfiguration {
    public function __construct(
        public readonly string $region,
        public readonly string $poolId,
        public readonly string $clientId,
    ) {
    }

    public function getIssuer(): string {
        return sprintf('https://cognito-idp.%s.amazonaws.com/%s_%s', $this->region, $this->region, $this->poolId);
    }

    public function getPublicKeysUrl(): string {
        return sprintf('https://cognito-idp.%s.amazonaws.com/%s_%s/.well-known/jwks.json', $this->region, $this->region, $this->poolId);
    }
}

Using this configuration and a HTTP client we can implement a key manager to download the keys:

<?php

declare(strict_types=1);

namespace Sam\JwtBlogPost;

use GuzzleHttp\ClientInterface;
use Jose\Component\Core\JWKSet;

class CognitoKeyManager {
    public function __construct(private ClientInterface $client, private CognitoConfiguration $configuration) {
    }

    public function getKeySet(): JWKSet {
        return JWKSet::createFromJson($this->retrieveKeys());
    }

    private function retrieveKeys(): string {
        // @todo These keys can be cached.
        return (string) $this->client->request('GET', $this->configuration->getPublicKeysUrl())->getBody();
    }
}

Decoding and Verifying Claims

From there, the token needs to be decoded, validated against the public keys and the claims in the token need to be validated, to ensure they are valid and originated from your specific Cognito user pool.

Using the key manager and our configuration, here is a working decoder based on Cognito's documentation about how tokens should be decoded and verified.

Key things to note are:

ID tokens and access tokens have slightly different means of validation (the aud and client_id claims need to be validated in each, respectively).
The token_use claim should be validated as either id or access respectively.
Other standard claims like iat, nbf, exp and iss should be validated in both. Cognito's Issuer convention is encapsulated in the CognitoConfiguration object.

<?php

namespace Sam\JwtBlogPost;

use Jose\Component\Checker\AlgorithmChecker;
use Jose\Component\Checker\AudienceChecker;
use Jose\Component\Checker\ClaimCheckerManager;
use Jose\Component\Checker\ExpirationTimeChecker;
use Jose\Component\Checker\HeaderCheckerManager;
use Jose\Component\Checker\IssuedAtChecker;
use Jose\Component\Checker\IssuerChecker;
use Jose\Component\Checker\NotBeforeChecker;
use Jose\Component\Core\AlgorithmManager;
use Jose\Component\Signature\Algorithm\RS256;
use Jose\Component\Signature\JWS;
use Jose\Component\Signature\JWSLoader;
use Jose\Component\Signature\JWSTokenSupport;
use Jose\Component\Signature\JWSVerifier;
use Jose\Component\Signature\Serializer\CompactSerializer;
use Jose\Component\Signature\Serializer\JWSSerializerManager;
use Sam\JwtBlogPost\Checkers\ClientIdChecker;
use Sam\JwtBlogPost\Checkers\TokenUseChecker;

/**
 * Load and verify Cognito tokens.
 *
 * Rules for verifying tokens are:
 *  - Verify that the token is not expired.
 *  - The aud claim in an ID token and the client_id claim in an access token should match the app client ID that was created in the Amazon Cognito user pool.
 *  - The issuer (iss) claim should match your user pool. For example, a user pool created in the us-east-1 Region will have the following iss value: https://cognito-idp.us-east-1.amazonaws.com/<userpoolID>.
 *  - Check the token_use claim.
 *    - If you are only accepting the access token in your web API operations, its value must be access.
 *    - If you are only using the ID token, its value must be id.
 *    - If you are using both ID and access tokens, the token_use claim must be either id or access.
 *
 * @see https://web-token.spomky-labs.com/advanced-topics-1/security-recommendations#loading-process
 * @see https://docs.aws.amazon.com/cognito/latest/developerguide/amazon-cognito-user-pools-using-tokens-verifying-a-jwt.html
 */
class CognitoJwtDecoder {
    public function __construct(private CognitoKeyManager $keyManager, private CognitoConfiguration $configuration) {
    }

    public function decodeIdToken(string $token): JWS {
        return $this->decodeAndValidate($token, [
            new AudienceChecker($this->configuration->clientId),
            new TokenUseChecker('id'),
        ], ['iss', 'aud', 'token_use']);
    }

    public function decodeAccessToken(string $token): JWS {
        return $this->decodeAndValidate($token, [
            new ClientIdChecker($this->configuration->clientId),
            new TokenUseChecker('access'),
        ], ['iss', 'client_id', 'token_use']);
    }

    /**
     * @throws \Jose\Component\Checker\InvalidClaimException
     * @throws \Jose\Component\Checker\MissingMandatoryClaimException
     * @throws \Exception
     */
    private function decodeAndValidate(string $token, array $claimChecks, array $mandatoryClaims): JWS {
        $headerChecker = new HeaderCheckerManager([new AlgorithmChecker(['RS256'])], [new JWSTokenSupport()]);
        $claimChecker = new ClaimCheckerManager(
            array_merge([
                new IssuedAtChecker(),
                new NotBeforeChecker(),
                new ExpirationTimeChecker(),
                new IssuerChecker([$this->configuration->getIssuer()]),
            ], $claimChecks)
        );

        $loader = new JWSLoader(new JWSSerializerManager([new CompactSerializer()]), new JWSVerifier(new AlgorithmManager([new RS256()])), $headerChecker);
        $jws = $loader->loadAndVerifyWithKeySet($token, $this->keyManager->getKeySet($token), $signature);

        $claims = json_decode($jws->getPayload(), true);
        $claimChecker->check($claims, $mandatoryClaims);

        return $jws;
    }

}

Pulling it all together

With all of these components in place, it's possible to pull together a proof of concept that can validate and decode Cognito JWTs:

<?php

require_once 'vendor/autoload.php';

[, $region, $poolId, $clientId, $type, $token] = $argv;

$config = new \Sam\JwtBlogPost\CognitoConfiguration($region, $poolId, $clientId);
$keyManager = new \Sam\JwtBlogPost\CognitoKeyManager(
    new \GuzzleHttp\Client(),
    $config,
);
$decoder = new \Sam\JwtBlogPost\CognitoJwtDecoder($keyManager, $config);

var_export($type === 'access' ? $decoder->decodeAccessToken($token) : $decoder->decodeIdToken($token));

Which can be invoked with:

php run.php us-east-2 POOL_ID CLIENT_ID TOKEN_TYPE TOKEN

Yielding a decoded token:

<?php
Jose\Component\Signature\JWS::__set_state(array(
'payload' => '{"sub":"420cd5cc-f537-4eab-a338-dc72f0b048e0","aud"...

It's worth mentioning, if you are using Symfony, there is a Symfony Bundle which will make some of the factories and services used in this blog post available from the container. In our application we decided instantiating these dependencies directly was preferable.

Programatically collecting statistics and quality metrics from Python packages

Sam — Fri, 20 Aug 2021 08:35:47 +0000

While in the process of comparing a large number of machine learning models as part of data science competitions run at Unearthed, in addition to analysing the predictive performance of models, we were interested in comparing some statistics and quality metrics for the actual source code files of each model.

Given there are a number of tools out there for analysing packages, the design goals were being able to collect of wide spectrum of information and something that could easily be invoked programatically (vs output from a CLI), to tie into our data science pipeline. With that in mind, the following options were evaluated:

In the end pylint and radon had the most promising and accessible feature set.

Radon had documented APIs for programatically accessing statistics in four categories: cyclomatic complexity, maintainability index, raw metrics and halstead metrics. An example of programatically collecting some statistics using the API is:

from radon.cli import Config
from radon.cli.harvest import MIHarvester


def measure_maintainability(source_dir):
    harvester = MIHarvester([source_dir], harvester_config())
    for path, raw_maintainability_statistics in harvester.results:
        print(raw_maintainability_statistics['mi'])


def harvester_config():
    return Config(exclude=None, ignore=None, order=SCORE, no_assert=False, show_closures=True, multi=4, by_function=False, min='A', max='F', include_ipynb=False)

Pylint lacked documentation on programatically collecting information from a source directory, but examining the entrypoint to the CLI command lead the following code snippet:

from pylint.lint import Run


def lint_directory(source_dir):
    buffer = io.StringIO()
    with redirect_stdout(buffer):
        try:
            Run(['--output-format=json', '--disable=' + ','.join(disabled_checks), source_dir])
        except:
            pass
    lint_results = json.loads(buffer.getvalue().replace("\n", ""))
    print(lint_results)

After collating each of the metrics, the response ended up looking like the following JSON payload:

{
  "aggregated_analysis": {
    "complexity": {
      "min": 1.0,
      "max": 5.0,
      "mean": 1.7619047619047619
    },
    "maintainability_index": {
      "min": 42.44120650814055,
      "max": 73.1183133154694,
      "mean": 61.16721652696456
    },
    "code_statistics": {
      "loc": 940,
      "lloc": 521,
      "sloc": 566,
      "comments": 183,
      "multi": 31,
      "blank": 186,
      "single_comments": 157
    },
    "halstead_metrics": {
      "h1": {
        "min": 1.0,
        "max": 15.0,
        "mean": 8.0
      },
      "h2": {
        "min": 3.0,
        "max": 137.0,
        "mean": 52.0
      },
      "N1": 110.0,
      "N2": 215.0,
      "vocabulary": {
        "min": 4.0,
        "max": 152.0,
        "mean": 60.0
      },
      "length": 325.0,
      "calculated_length": {
        "min": 4.754887502163469,
        "max": 1031.03375429972,
        "mean": 374.5962139339611
      },
      "volume": {
        "min": 12.0,
        "max": 2101.89897889864,
        "mean": 748.9542971398513
      },
      "difficulty": {
        "min": 0.6666666666666666,
        "max": 10.510948905109489,
        "mean": 5.309205190592052
      },
      "effort": {
        "min": 8.0,
        "max": 22092.952770905413,
        "mean": 7577.510451793251
      },
      "time": 1262.9184086322082,
      "bugs": 0.7489542971398513
    },
    "linting": {
      "types": {
        "convention": 206,
        "warning": 55,
        "refactor": 12
      },
      "symbols": {
        "invalid-name": 85,
        "line-too-long": 53,
        "trailing-whitespace": 33,
        // ...
      }
    }
  },
  "file_analysis": {
    // The same metrics, but for each file in the package.
  }
}

Designing a Data Science oriented tabular data query DSL

Sam — Fri, 25 Jun 2021 07:33:16 +0000

I have been working on a number of services written in JavaScript, Python and PHP that organise, orchestrate and visualise Machine Learning workloads. One of the most common artefacts that feed in and out of these services is tabular data, which may come in a number of forms such as:

Raw data from some kind of sensor.
Model training datasets.
Predictions from an inference pipeline.

There are a number of useful thing a user or back-end service might expect to be able to accomplish with these kind of tabular data files. A few examples might be:

A user may want to paginate though the first N rows of a large file, to preview rows.
A service may want to generate quality metrics from a series of predictions made by a machine learning model.
A service may want to split a dataset into two buckets for training and validation.

Since we're dealing with potentially multi-gigabyte files, it's impractical for clients and services to download, filter, then transfer large volumes of data over the network, so most services will reference file URIs contained in some mutually accessible store. When communicating with URIs the problem then becomes:

In addition to a URI, how do clients and services communicate which rows and columns within a data file are meaningful for a particular task?

Additional design goals were:

It needed to be fast, for large datasets.
Queries needed to be authored in PHP and JavaScript applications and then executed in a Python environment.
It needed a rich syntax, so we could effectively paginate, slice and select data for a variety of use cases.

With that in mind, I started to evaluate a number of projects that implemented some kind of query language that could be sent along with our data file URIs, a couple that came up:

CSVkit - Showed some promise, but the query features depended on importing the data into a SQLite database that locked up and fell over as soon as large datasets were involved.
q - Also had a SQLite dependency and failed to deal with large amounts of data.
Miller - Looked interesting but also seemed like a complex syntax and more focused on being CLI tool than integrating into a Python runtime.

One of the tools we were already using which was both extremely feature rich and worked with extremely large datasets was from the pandas module: pandas.DataFrame. I found that almost all our needs would be met if we could represent a series of operations against a DataFrame and pass those from service to service.

In it was settled that queries would be represented as a string containing JSON, when decoded would be an array of operations that could be executed against a DataFrame. Some examples of JSON strings and the corresponding operation

Select a column named "foo":

[
  {
    "method": "get",
    "args": ["foo"]
  }
]

df.get("foo")

Select the 5th numerically indexed column:

[
  {
    "index": "iloc",
    "keys": [":",5]
  }
]

df.iloc[:, 5]

Select column "bar", then select the first 10 rows:

[
  {
    "method": "get",
    "args": ["bar"]
  },
  {
    "method": "head",
    "args": [10]
  }
]

df.get("bar").head(10)

In the end, the code for translating these JSON strings into actual transformations on a DataFrame was quite simple:

import json


def apply_query(data_frame, query):
    parsed_query = json.loads(query)

    query_applied_object = data_frame

    for operation in parsed_query:
        if "method" in operation:
            query_applied_object = getattr(query_applied_object, operation["method"])(*operation["args"])

        if "index" in operation:
            index = getattr(query_applied_object, operation["index"])
            # Support representing a single colon string, ":" in the query object as a null slice,
            # this could be further extended to support more complex slice objects like "1:2" if
            # required in the future.
            keys = list(map(lambda accessor: slice(None) if accessor == ":" else accessor, operation["keys"]))
            query_applied_object = type(index).__getitem__(index, tuple(keys))

    return query_applied_object

With the format defined and our Python runtime recognising and executing it, it was then straightforward to integrate queries into our Symfony based PHP app as a value object:

<?php

declare(strict_types=1);

namespace App\Model\DataFile\Query;

/**
 * Represents a query against a data file.
 * 
 * Examples:
 * @code
 * Query::columnByName('foo');
 * Query::columnByIndex(3);
 * @endcode
 */
final class Query implements \Stringable {
    const NULL_SLICE = ':';

    private function __construct(private array $query) {
    }

    public static function columnByName(string $columnName): static {
        return new static([
            [
                'method' => 'get',
                'args' => [
                    $columnName,
                ],
            ],
        ]);
    }

    public static function columnByIndex(int $columnIndex): static {
        return new static([
            [
                'index' => 'iloc',
                'keys' => [
                    static::NULL_SLICE,
                    $columnIndex,
                ],
            ],
        ]);
    }

    public function __toString() {
        return json_encode($this->query);
    }

    public static function fromString(string $string): static {
        return new static(json_decode($string, true));
    }
}

Crude Python tree shaking for squeezing into AWS Lambda package size limits

Sam — Wed, 23 Jun 2021 23:19:03 +0000

I have been working on a service that manages a number of Machine Learning workloads that is deployed to AWS Lambda. One of the issues I encountered deploying this service is our dependency on a number of popular ML related Python modules such as pandas and scikit-learn and exceeding the maximum size thresholds imposed by AWS. AWS enforces a 250 MB hard limit on Lambda packages and scikit-learn alone is 100 MB unzipped. Deploying yields:

ServerlessError: An error occurred: Resource handler returned message: "Unzipped size must be smaller than 262144000 bytes (Service: Lambda, Status Code: 400)"

There doesn't seem to be many options for slimming down dependencies, I was hoping to find subtree splits of the submodules that were relevant to our project, but came up short. I had already configured the serverless-python-requirements plugin to slim down packages:

custom:
  pythonRequirements:
    slim: true
    strip: false

Note: the strip: false option was required to prevent an error during execution which manifested as:

Runtime.ImportModuleError: Unable to import module 'handler': /var/task/scipy/linalg/_fblas.cpython-38-x86_64-linux-gnu.so: ELF load command address/offset not properly aligned

One option to reduce the footprint of dependencies is to implement some form of tree shaking, to remove chunks of code we're not actually using, the tricky part would be to identify which parts those are. In our case, we had already 100% test coverage, so we were able to use this coverage and run a report against our dependency folder (instead of our src folder) to find out which parts of our dependencies we were actually using:

poetry run pytest --cov-report=html\
 --cov=/path/to/site-packages

Running this report indicates about %6 "coverage" of our dependencies:

From here, it was a simple matter of taking a snapshot of files with 0% coverage and removing them from our artefact. To accomplish this, I found the serverless-scriptable-plugin was required to remove dependencies added by the serverless-python-requirements plugin.

serverless.yml:

plugins:
  - serverless-python-requirements
  - serverless-scriptable-plugin
...
custom:
  pythonRequirements:
    slim: true
    strip: false
  scriptable:
    hooks:
      after:package:createDeploymentArtifacts:
        - ./shake.sh

shake.sh

#/bin/sh
filelist="_distutils_hack/__init__.py
_distutils_hack/override.py
...
wheel/util.py
wheel/vendored/packaging/_typing.py
wheel/vendored/packaging/tags.py
wheel/wheelfile.py"

for file in $filelist
do
  zip --delete ./.serverless/data-science-api.zip $file
done

In the end 28,000 unnecessary files were removed and the artefact was reduced from 307 MB to 236 MB, unblocking deployment.