はじめに

こんにちは。NTTデータ先端技術の@tsuchidahrkです。
本記事は、データ活用基盤を作ってみた連載記事のその8です。
本シリーズの取り組みの内容についてはその１(構成シナリオ)をご覧ください。
前の記事は、その７(構築・IAM)からご覧ください。
本記事では、今回作成したデータ活用基盤データ収集処理の設計、構築方法、データ収集処理の実装について紹介します。

目次

データ収集処理設計
データ収集処理の流れ
構築方法
データ収集処理の実装
まとめ

データ収集処理の設計

データ活用基盤を構築する際に欠かせないデータ収集処理の設計について紹介します。
本システムでは、その1の機能・非機能要望を満たせるように設計しました。
データ収集処理に関連する機能・非機能要望は以下になります。

	項目	要望
機能	-	月次ジョブLambdaで日本郵便からzipファイルダウンロードしてS3に保存する
非機能	可用性	新構成のデータ活用基盤側は全てマネージドサービスを使用すること
	性能/拡張性	SmallStart時点で具体的な性能目標は定義しない。ただし、今後の拡大かに備えて、スケール可能なリソースを利用する
	運用・保守性	営業時間帯 10:30～18:00にオンライン作業ができること。

上記の機能、非機能要望から以下のように設計しました。

項目	設計内容	理由
収集サービス	収集については、Lambdaを利用。月次ジョブの実行には、EventBridgeとStepFunctionを利用。	収集については、機能要望で指定されていたため。月次ジョブの実行については、マネージドサービスを利用する必要があったため。
収集タイミング	毎月1日の0:00に収集。タイムゾーンはAsia/Tokyo	オンライン作業に影響を与えない時間帯を選んだため。
収集先	日本郵便のサイト	機能要望で指定されていたため。
収集対象	都道府県一覧の全国一括データ ken_all.zip	機能要望で指定されていたため。
収集対象の格納先	S3バケット	機能要望で指定されていたため。

データ収集処理の流れ

構成図

手順

毎月1日の0:00にEventBridge Schedulerが起動し、Step Functionのワークフローを動作させます。
ワークフロージョブのLambdaを起動し、処理を実行します。
Lambdaは、郵便局サイトにアクセスして、「ken_all」というファイル名のZIPファイルをローカルストレージにダウンロードします。その後、S3上の加工前バケットにダウンロードしたZIPファイルをLambdaのローカルストレージからアップロードします。
後続のデータ加工処理にワークフロージョブが遷移します。

構築方法

今回作成しているAWSリソースはTerraformを用いたInfrastructure as Code（IaC）で作成しています。以下に作成したTerraform構成ファイル（jobs.tf）の内容について記載します。

データ収集処理

Lambdaリソースの作成

Lambdaリソースで実行するスクリプトが配置されたS3バケットとファイル名を記載しています。
また、スクリプトの動作環境となるランタイムのバージョン設定とスクリプト内のハンドラー登録を行うよう定義しています。

jobs.tf

resource "aws_lambda_function" "biwa-prod-lambda-function-zip-001" {
  function_name = "biwa-prod-lambda-function-zip-001"
  role = aws_iam_role.biwa-prod-iam-role-lambda-execution-001.arn

  s3_bucket = "biwa-prd-s3-bucket-scripts-001"
  s3_key = "lambda_function.zip"

  runtime = "python3.9"
  handler = "lambda_function.lambda_handler"
}

月次ジョブ

EventBridge Schedulerリソースの作成

スケジュール設定と実行ターゲットの設定をしています。
スケジュール設定では、タイムゾーンに東京を指定し、cron形式で毎月1日の0:00にジョブが実行するよう定義しています。
また、実行ターゲットの設定では、Step Functions WorkFlowを実行対象として定義しています。

jobs.tf

resource "aws_scheduler_schedule" "biwa-prod-EventBridge-scheduler-monthly-001" {
  name = "biwa-prod-EventBridge-scheduler-monthly-001"

  flexible_time_window {
    mode = "OFF"
  }

  schedule_expression          = "cron(0 1 * ? *)"
  schedule_expression_timezone = "Asia/Tokyo"

  target {
    arn = "arn:aws:states:ap-northeast-1:125336375984:stateMachine:biwa-prod-sfn-statemachine-zip-001"
    role_arn = aws_iam_role.biwa-prod-iam-role-EventBridge-execution-001.arn
  }
}

Step Functionsリソースの作成

リソース作成時に、データ収集処理を行うLambda関数の実行と後続処理のデータ加工処理のジョブも定義しています。
データ加工処理のジョブについては、データ活用基盤を作ってみたその10で紹介します。

jobs.tf

resource "aws_sfn_state_machine" "biwa-prod-sfn-statemachine-zip-001" {
  name = "biwa-prod-sfn-statemachine-zip-001"
  role_arn = aws_iam_role.biwa-prod-iam-role-stepFunctions-execution-001.arn

  definition = <<EOF
  {
      "StartAt": "RunExtractionJob",
      "States": {
        "RunExtractionJob": {
          "Type": "Task",
          "Resource": "${aws_lambda_function.biwa-prod-lambda-function-zip-001.arn}",
          "Next": "RunPreprocessingJob"
        },
        "RunPreprocessingJob": {
          "Type": "Task",
          "Resource": "arn:aws:states:::glue:startJobRun.sync",
          "Parameters": {
            "JobName": "biwa-prod-glue-job-zip-001"
          },
          "Retry": [
            {
              "ErrorEquals": ["States.ALL"],
              "IntervalSeconds": 1,
              "MaxAttempts": 3,
              "BackoffRate": 2.0
            }
          ],
          "Next": "WaitForPreprocessingJob"
        },
        "WaitForPreprocessingJob": {
          "Type": "Wait",
          "Seconds": 60,
          "Next" : "RunCrawler"
        },
        "RunCrawler": {
          "Type": "Task",
          "Resource": "arn:aws:states:::aws-sdk:glue:startCrawler",
          "Parameters": {
            "Name": "biwa-prod-glue-crawler-analytics-001"
          },
          "End" : true
        }
      }
  }
  EOF
}

データ収集処理の実装

データ収集処理の流れで説明した手順3の郵便局サイトからZIPファイルを収集しS3バケットへ格納する処理の実装が以下になります。

lambda_function.py

import urllib
import boto3
from botocore.config import Config
import botocore.exceptions
import os
import logging

URL = 'https://www.post.japanpost.jp/zipcode/dl/kogaki/zip/ken_all.zip'
FILENAME = 'ken_all.zip'
DOWNLOAD_PATH = os.path.join('/tmp', FILENAME)

AWS_CONFIG = Config(
    connect_timeout = 120,
    read_timeout = 120
)

S3 = boto3.resource('s3', config=AWS_CONFIG)
BUCKET = S3.Bucket('biwa-prod-s3-bucket-datastorage-001')

def download_zipcode(url, download_path):
    """郵便局データファイルのダウンロード

    郵便局サイトにアクセスし、郵便番号データファイルをダウンロードする。

    :param String url: 郵便局サイトのURL
    :param String download_path: ファイルを書き込むパス
    :return: 戻り値なし
    :raises: URLError: URLが存在しない
    """
    try:
        urllib.request.urlretrieve(url, download_path)
        logging.info('file download success!')
        return None
    except urllib.error.URLError as e:
         print(e)

def upload_s3(bucket, download_path, filename):
    """S3へ郵便局データファイルのアップロード

    S3にアクセスし、ローカルから郵便番号データファイルをアップロードする。

    :param boto3.resources.factory.s3.Bucket Bucket: 郵便局サイトのURL
    :param String download_path: ローカルの郵便番号データファイルのパス
    :param String filename: 郵便番号データファイルのファイル名
    :return: 戻り値なし
    :raises: S3UploadFailedError: S3へのファイルアップロードに失敗
    """
    try:
        bucket.upload_file(download_path, filename)
        logging.info('file upload success!')
        return None
    except boto3.exceptions.S3UploadFailedError as e:
            print(e)

def lambda_handler(event, context):
    """郵便局サイトから郵便番号データファイルをダウンロードし、S3へファイルをアップロードする。

    メイン処理

    :param Dict event: テストの値
    :param awslambdaric.lambda_context.LambdaContext context: ランタイム情報
    :return: 戻り値なし
    :raises: 例外処理なし
    """
    download_zipcode(URL, DOWNLOAD_PATH)
    upload_s3(BUCKET, DOWNLOAD_PATH, FILENAME)

    if os.path.isfile(DOWNLOAD_PATH):
        os.remove(DOWNLOAD_PATH)
        logging.info('file remove success!')
        return None
    
    return None

上記、データ収集処理のプログラムは以下の流れで動作します。

lambda_handlerメソッドからdownload_zipメソッドが呼び出されます。download_zipメソッドより、urllib.request.urlretrieve関数を用いて郵便局サイトにアクセスし、URLリンクへのアクセスができれば郵便番号データファイルをLambdaのローカルストレージにダウンロードします。
lambda_handlerメソッドに戻り、upload_s3メソッドが呼び出されます。upload_s3メソッドより、指定されたS3バケットへLambdaのローカルストレージからアップロードします。
Lambdaのローカルストレージにある郵便番号データファイルを削除します。

まとめ

今回はデータ収集処理の設計や、TerraformのIacを用いてLambdaリソースを作成する方法について紹介しました。
Lambdaを利用したことやAWS上のリソース間を連携させて処理を実装したことは無かったので、AWS上のマネージドサービスを組み合わせて処理を作る楽しさを知ることが出来ました。
今後は複数のデータソースから様々な方法でデータ収集処理を実現出来るようにしていきたいです。

次の記事は、その９(構築・データ蓄積-S3)です。
最後までお読みいただきありがとうございました。