Code NodeワークフローでAthena接続用の外部パッケージをn8nに追加する

version: '3.8'
services:
  n8n:
    image: n8nio/n8n:latest
    container_name: n8n
    restart: unless-stopped
    ports:
      - "5678:5678"
    environment:
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
      - GENERIC_TIMEZONE=${GENERIC_TIMEZONE}
    volumes:
      - ./n8n_data:/home/node/.n8n

FROM n8nio/n8n:latest
USER root
RUN npm install -g @aws-sdk/client-athena
USER node

$ docker build -t n8n-custom-v1.0.1 .

export NODE_FUNCTION_ALLOW_EXTERNAL=@aws-sdk/client-athena

n8n:
    #image: n8nio/n8n:latest
    image: {{ビルドしたDocker Image名}}
    restart: always
    ports:
      - "127.0.0.1:5678:5678"
    environment:
      - DB_TYPE=postgresdb
      〜〜〜〜〜
      略
      〜〜〜〜〜
      - NODE_FUNCTION_ALLOW_EXTERNAL=@aws-sdk/client-athena
    volumes:
      - n8n_storage:/home/node/.n8n
    depends_on:
      postgres:
        condition: service_healthy

$ docker-compose up -d

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowAthenaDataSourceRead",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::<<your alb log path>>",
                "arn:aws:s3:::<<your alb log path>>/*"
            ]
        },
        {
            "Sid": "AllowAthenaQueryResultWrite",
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListMultipartUploadParts",
                "s3:AbortMultipartUpload",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::<your-athena-query-results-bucket>",
                "arn:aws:s3:::<your-athena-query-results-bucket>/*"
            ]
        },
        {
            "Sid": "AllowAthenaAndGlueAPI",
            "Effect": "Allow",
            "Action": [
                "athena:StartQueryExecution",
                "athena:GetQueryExecution",
                "athena:GetQueryResults",
                "athena:StopQueryExecution",
                "glue:GetTable",
                "glue:GetDatabase",
                "glue:GetPartitions"
            ],
            "Resource": "*"
        }
    ]
}

// AWS SDK for JavaScript v3 のAthenaクライアントをインポート
const {
  AthenaClient,
  StartQueryExecutionCommand,
  GetQueryExecutionCommand,
  GetQueryResultsCommand
} = require("@aws-sdk/client-athena");

// --- 設定項目 ---
const AWS_REGION = 'ap-northeast-1'; // 例: 'ap-northeast-1' (東京リージョン)
const ATHENA_DATABASE = '＜＜your athena database name＞＞'; // Athenaのデータベース名
const ATHENA_OUTPUT_LOCATION = 's3://<<S3 path to output query results>>'; // クエリ結果の出力先S3パス

// 実行したいSQLクエリ
const SQL_QUERY = 'SELECT * FROM "<<your athena table>>" LIMIT 10;';
// --- 設定項目ここまで ---

// 非同期処理を実行するためのメイン関数
// n8nの実行コンテキスト(this)を引き継ぐために .call(this) で呼び出します。
async function main() {
  try {
    // Athenaクライアントの初期化
    const athenaClient = new AthenaClient({
      region: AWS_REGION
    });

    // 2. クエリ実行を開始
    const startQueryExecutionCommand = new StartQueryExecutionCommand({
      QueryString: SQL_QUERY,
      QueryExecutionContext: {
        Database: ATHENA_DATABASE,
      },
      ResultConfiguration: {
        OutputLocation: ATHENA_OUTPUT_LOCATION,
      },
    });

    const { QueryExecutionId } = await athenaClient.send(startQueryExecutionCommand);
    console.log(`Query Execution ID: ${QueryExecutionId}`);

    if (!QueryExecutionId) {
        throw new Error('Query Execution ID not returned.');
    }

    // 3. クエリの実行完了をポーリングで待機
    let queryState = '';
    do {
      // 3秒待機
      await new Promise(resolve => setTimeout(resolve, 3000));

      const getQueryExecutionCommand = new GetQueryExecutionCommand({ QueryExecutionId });
      const queryExecutionResponse = await athenaClient.send(getQueryExecutionCommand);
      queryState = queryExecutionResponse.QueryExecution.Status.State;
      console.log(`Current query state: ${queryState}`);

      if (['FAILED', 'CANCELLED'].includes(queryState)) {
        throw new Error(`Query ${queryState}. Reason: ${queryExecutionResponse.QueryExecution.Status.StateChangeReason}`);
      }
    } while (queryState !== 'SUCCEEDED');

    // 4. クエリ結果を取得
    const getQueryResultsCommand = new GetQueryResultsCommand({ QueryExecutionId });
    const queryResults = await athenaClient.send(getQueryResultsCommand);

    // 5. 結果をn8nが扱いやすいJSON形式に整形
    const resultSet = queryResults.ResultSet;
    const headers = resultSet.Rows[0].Data.map(header => header.VarCharValue);
    const rows = resultSet.Rows.slice(1).map(row => {
      const rowData = {};
      row.Data.forEach((item, index) => {
        rowData[headers[index]] = item.VarCharValue;
      });
      return rowData;
    });

    // n8nの出力形式に合わせてデータを返す
    return [{ json: { data: rows } }];

  } catch (error) {
    console.error(error);
    // エラーが発生した場合もn8nに情報を返す
    return [{ json: { error: error.message, stack: error.stack } }];
  }
}

return main.call(this);

SELECT substr(time, 1, 10) AS req_date, count(request_url) as cnt 
FROM "hermes"."alb_access_logs" 
Where time between '2025-07-01' and '2025-07-31'
GROUP BY substr(time, 1, 10), substr(time, 1, 10)
order by req_date asc
;