activity-spark-aries.js

import { SparkActivity } from 'aries-data';
import AWS from 'aws-sdk';
import moment from 'moment';

export default class AcmeDailyAnalysis extends SparkActivity {
    static props = {
        name: require('../package.json').name,
        version: require('../package.json').version,
    };

    @singleS3StreamOutput()
    async onTask(activityTask, config) {
        // Create a dataframe from s3 files. 
        const dataframe = this.sqlContext.read().jsonSync(this.s3Path());

        // Spark query
        const ctx = df.select("context.userAgent")
            .groupBy('userAgent')
            .count()
            .sort('count')
 
        // Run the job and return the results.
        return ctx.collectSync();
    }

    // Return a s3 path for yesterdays data.
    s3Path() {
    const yesterday = moment.utc().subtract(1, 'day').format('YYYY/MM/DD');
        return `s3n://acme-clickstream/data/${yesterday}/*`;
    }
};