AWS LambdaとFivetranを連携して、長州力のツイートをSnowflakeにロードするぞタココラ　井長州力

{
    "agent" : "<function_connector_name>/<external_id>/<schema>",
    "state": {
        "cursor": "2018-01-01T00:00:00Z"
    },
    "secrets": {
        "apiToken": "abcdefghijklmnopqrstuvwxyz_0123456789"
    }
}

{
    "state": {
        "transaction": "2018-01-02T00:00:00Z",
        "campaign": "2018-01-02T00:00:01Z"
    },
    "insert": {
        "transaction": [
            {"id":1, "amount": 100},
            {"id":2, "amount": 50}
        ],
        "campaign": [
            {"id":101, "name": "Christmas"},
            {"id":102, "name": "New Year"}
        ]
    },
    "delete": {
        "transaction": [
            {"id":3},
            {"id":4}
        ],
        "campaign": [
            {"id":103},
            {"id":104}
        ]
    },
    "schema" : {
        "transaction": {
            "primary_key": ["id"]
        },
        "campaign": {
            "primary_key": ["id"]
        }
    },
    "hasMore" : true
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "InvokePermission",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": "*"
        }
    ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::834469178297:root"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "<FivetranのExternal ID>"
        }
      }
    },
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

const https = require('https');

exports.handler = (request, context, callback) => {
    // Twitter APIのトークンを取得する
    console.log('Getting access token from twitter...')
    let post = https.request({
        method: 'POST',
        hostname: 'api.twitter.com',
        path: '/oauth2/token',
        headers: {
            'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
            'Content-Length': 'grant_type=client_credentials'.length,
            'Authorization': 'Basic ' + Buffer.from(`${request.secrets.consumerKey}:${request.secrets.consumerSecret}`, 'utf8').toString('base64')
        }
    }, postRes => {
        var reply = ''
        postRes.setEncoding('utf8')
        postRes.on('data', chunk => reply += chunk)
        postRes.on('end', () => withAuth(JSON.parse(reply)))
    })
    post.on('error', e => console.log(`Problem with request ${e}`))
    post.write('grant_type=client_credentials')
    post.end()

    // トークンを取得したら、それを使って長州力のタイムラインをリクエストする
    function withAuth(auth) {
        console.log('...got access token from twitter, fetching tweets...')
        var path = '/1.1/statuses/user_timeline.json?screen_name=rikichannel1203'
        // これが初回リクエストではない場合は、リクエストパスに since_id={since_id} を追加する
        if (request.state.since_id != null) {
            console.log(`...getting new tweets since id ${request.state.since_id}...`)
            path += '&since_id=' + request.state.since_id
        }
        // 直近のツイートを取得する
        let get = https.get({
            hostname: 'api.twitter.com',
            path: path,
            headers: {
                'Authorization': 'Bearer ' + auth.access_token
            }
        }, getRes => {
            var reply = ''

            getRes.on('data', chunk => reply += chunk)
            getRes.on('end', () => withTweets(JSON.parse(reply)))
        })
    }

    // 長州力のタイムラインを取得したら、それを使ってDWHにデータをアップサートする
    function withTweets(timeline) {
        console.log(`...got ${timeline.length} tweets, sending to Fivetran.`)
        // 最新のIDを記録しておくことで、今後の更新を増分更新にする
        let since_id = null
        // Twitter APIからのレスポンスをフラットテーブルの形に再フォーマットする
        let tweets = []
        let users = []
        for (let t of timeline) {
            // 最初のIDを記録する
            if (since_id == null) {
                since_id = t.id
            }
            // すべてのツイートを追加する
            tweets.push({
                id: t.id,
                user_id: t.user.id,
                text: t.text
            })
            // ユーザー情報を追加する
            // 重複処理はFivetran側で処理してくれるので、ここで考える必要はない
            users.push({
                id: t.user.id,
                name: t.user.name,
                screen_name: t.user.screen_name
            })
        }
        // JSONのレスポンスをFivetranに送信する
        let resulut = ({
            // 最新のIDを記録しているため、リクエストは増分更新となる
            state: {
                since_id: since_id == null ? request.state.since_id : since_id
            },
            // Fivetranはこれらの主キーを使用してマージを実行する
            // そのため、同じレコードを2回送信しても、DWHには1つ分しか送られない
            schema : {
                tweets : {
                    primary_key : ['id']
                },
                users: {
                    primary_key: ['id']
                }
            },
            // 下記のレコードをDWHに挿入する
            insert: {
                tweets: tweets,
                users: users
            },
            // これがtrueの場合、Fivetranは再び関数を実行し、再度データを取得する
            // 大量のデータ取得に役立つ
            hasMore : false
        });

        callback(null, resulut);
    }
}

{
    "consumerKey": "<Twitter APIのアクセスキー>",
    "consumerSecret": "<Twitter APIのアクセスシークレット>"
}