nodeでバッチスクリプトを書くときの道具箱

お久しぶりです。
業務をしているとちょっとしたバッチスクリプトを書くことがあると思います。最近たまたまnodeでCSVを使ったバッチ処理を書くことがあったので、その際のtipsをまとめてみます。 nodeは最近標準ライブラリが少しずつ充実してきましたが、普段からゴリゴリnode書いてない身からするとGoやRubyといった標準ライブラリやノウハウが浸透している言語の方がやりやすい部分はあるなと感じた今日この頃

紹介するtips

CSVから行を逐次読み込みする
コマンドライン引数を取得する
タイムスタンプのついた構造化ログを吐く
非同期処理の並列数を制限しつつ並列で実行する
jest, vitestを使わずにテストコードを書く

CSVから行を逐次読み込みする

バッチ処理などでは巨大なファイルを扱うことも多いでしょう。
ファイルを扱う場合は、ファイル全体を読み込むのではなく1行ずつ読み込みながら処理を進めていくことで少ないメモリ使用量でサイズの大きなファイルも扱うことができます。
そのため、あらかじめファイルサイズが限定されている場合を除いて出来るだけ逐次行を読み込んで処理する書き方をすることが一般的でしょう。

nodeではcsvというライブラリを使うことでCSVの生成・変換・読み込み・書き出しをすることができます。 CSVをパースする方法は同期的にパースする方法やcallback関数を使う方法もありますが、今回はstreamを使う方法を紹介します。 fs.createReadStrem を使ってファイルのストリームを取得しpipeでparserを渡してあげることで、CSVの一行ずつイテレートする AsyncIterator を取得することができます。

import * as fs from 'node:fs';
import { parse } from 'csv';


const fileStream = fs.createReadStream('items.csv');

const rows = fileStream.pipe(parse({ trim: true, skip_empty_lines: true }));

for await (const row of rows) {
    console.log(row); // ['xxxx', 'yyyy', 'zzzz']
}

CSV Parse - Async iterator API

コマンドライン引数を取得する

これまでも process.argv を使うことでコマンドライン引数を扱うことができましたが、最近のバージョンでより便利なutil関数が標準ライブラリに追加されました。 util.parseArgs はv18.3でexperimentalとして追加されv20.0から正式なAPIになりました。

import * as util from 'node:util';

const { values, positionals } = util.parseArgs({
    allowPositionals: true, // デフォルトでは、オプション指定のない引数は指定できない
    options: {
        env: {
            type: 'string',
            short: 'e',
        },
        target: {
            type: 'string',
            short: 't',
            multiple: true,
        },
    },
});

console.log(values)
console.log(positionals)

$ node parse-args.mjs items.csv -e local -t foo --target bar
> [Object: null prototype] { env: 'local', target: [ 'foo', 'bar' ] }
> [ 'items.csv' ]

Util | Node.js v21.6.2 Documentation

タイムスタンプのついた構造化ログを吐く

バッチ処理ではいつどこまで処理が進んだか・いつどんなエラーが出たかを記録しておくことが重要です。
その際には、ログにタイムスタンプが必要になりますし、またログが構造化されていると後々解析するのに便利です。
また、スクリプトを実行中ログをターミナルに出力しつつもファイルにも保存しておきたいということもあるでしょう。今回は、pinoというライブラリを使ってこの辺りを実現してみます。

import { pino, stdTimeFunctions } from 'pino';

const transport = pino.transport({
  dedupe: true,
  targets: [
    {
      level: 'info',
      target: 'pino/file',
      options: { destination: '1' }, // stdout
    },
    {
      level: 'error',
      target: 'pino/file',
      options: { destination: '2' }, // stderr
    },
    {
      level: 'info',
      target: 'pino/file',
      options: { destination: './logs/info.log', mkdir: true },
    },
    {
      level: 'error',
      target: 'pino/file',
      options: { destination: './logs/error.log', mkdir: true },
    }
  ]
});

const logger = pino({
  timestamp: stdTimeFunctions.isoTime,
}, transport);

logger.info('info'); // written to stdout and info.log
logger.warn('warn'); // written to stdout nad info.log
logger.error('error'); // written to stderr and error.log
logger.fatal('fatal'); // written to stderr and error.log

上記のようにするとターミナルとそれぞれのファイルに以下のようなログを出力することができます。

// info.log
{"level":30,"time":"2023-04-23T03:23:47.960Z","pid":79891,"hostname":"Ayatos-MBP","id":1,"msg":"info"}
{"level":50,"time":"2023-04-23T03:23:47.960Z","pid":79891,"hostname":"Ayatos-MBP","id":3,"msg":"error"}

// error.log
{"level":40,"time":"2023-04-23T03:23:47.960Z","pid":79891,"hostname":"Ayatos-MBP","id":2,"msg":"warn"}
{"level":60,"time":"2023-04-23T03:23:47.960Z","pid":79891,"hostname":"Ayatos-MBP","id":4,"msg":"fatal"}

また、構造化されたログは人間には読みづらいため開発環境では pino-pretty を使うとヒューマンリーダブルなログをターミナルに表示することができます。

GitHub - pinojs/pino: 🌲 super fast, all natural json logger

非同期処理の並列数を制限しつつ並列で実行する

バッチスクリプトでは、多くのケースでhttpリクエストやDBアクセスなどのIO処理を利用するでしょう。
その際、出来るだけ効率的に処理するために処理をasync関数としてIOを並列化して処理することが考えられます。しかし、単に全ての処理を非同期実行して Promise.all するようなやり方ではリクエストの並列数をコントロールすることができません。
この解決には複数の方法が考えられ、タスクをチャンクしチャンクごとに Promise.all するようなやりかたなども考えられますが、今回のような AsyncIterator を使うような場合はchunkしづらいため別の方法を取ります。

今回は、iteratorを消費するワーカーを複数作ることで、並列に実行しつつ並列数を制御します。
このやり方の場合、 for await ... of を使うことで簡単に AsyncIterator に対応することができますね。

const iterator = Array.from('abcdefghi').values();

async function doWork(iterator) {
  for await (const value of iterator) {
    await setTimeout(1000);
    console.log(value);
  }
}

// イテレータを消費するワーカーを3つ作る
const workers = new Array(3).fill(iterator).map(doWork);

// 全てのワーカーの処理完了を待つ
await Promise.allSettled(workers);

a
b
c
# 1秒後
d
e
f
# ...

Run Concurrent Tasks With a Limit Using Pure JavaScript - Maxim Orlov

jest, vitestを使わずにテストコードを書く

書き捨てのスクリプトだがテストは書いておきたいという場合にわざわざTesting Libraryを導入して設定するのは面倒ですよね。
v18からexperimentalで導入されていたnode標準のTest Runnerがv20で無事stableになりました。
サクッとテストを書きたい場合にライブラリをインストールする必要がないのでありがたいですね。

非同期処理を含むテストも書けますし、テストの構造化もできます。describe,it を使って書き方や before , afterEach といったフックもできるみたいなので、ちょっとテスト書いて動作確認したいといった場合にライブラリが必要になる場面がかなり減りそうです。

import { test, describe, it } from 'node:test'
import * as assert from 'node:assert';


test('first test', (t) => {
  assert.strictEqual(1, 1);
});

test('asynchronous passing test', async (t) => {
  assert.strictEqual(1, 1);
});

test('top level test', async (t) => {
  await t.test('subtest', (t) => {
    assert.strictEqual(1, 1);
  });
}); 

describe('A thing', () => {
  it('should work', () => {
    assert.strictEqual(1, 1);
  });
});

実行結果もそこそこ見やすそうでいい感じです。

Test runner | Node.js v21.6.2 Documentation

感想

node自体は慣れているつもりでしたが、普段書かないような処理を書こうと思うと意外に標準ライブラリの使い方やライブラリの選定が必要になりますね。
近年nodeの標準ライブラリが便利になってきているのでさらに今後に期待です。