Symfoware

Node.js FileStreamの読み込みを一時停止しつつ、一括登録を実行する

郵便番号データをTedious BulkLoadで一括登録してみました。
Node.js Tedious でBulkLoadを使用して郵便番号データを一括登録

このときは郵便番号約12万件をすべて読みこんで一括登録しましたが、
登録するデータ件数が増えた場合を考慮し、1万件毎にBulkLoadするよう修正してみます。

最初のサンプル

1万件読み込んだらBulkLoadを実行すればよいだろうと修正したソースがこちら。

const fs = require('fs');
const readline = require('readline')
const { Connection, TYPES } = require('tedious')
// 設定に従いデータベースへ接続
function create_connection(config) {
    
    const connection = new Connection(config)
    // Promiseをnewした時点で引数のfunctionが実行される
    const p = new Promise(function(resolve, reject) {
        connection.on('connect', err => {
            if (err) {
                reject(err)
            } else {
                resolve(connection)
            }
        });
        connection.connect()
    });
    return p
    
}
// BulkLoadの実行
function execBulkLoad(connection, rows) {
    const p = new Promise(function(resolve, reject) {
        //　BulkLoad用の設定
        const options = {}
        const bulkLoad = connection.newBulkLoad('postal_code', options, function (error, rowCount) {
            resolve('inserted ' + rowCount + ' rows');
        })
        bulkLoad.addColumn('code', TYPES.Char, { nullable: false })
        bulkLoad.addColumn('address', TYPES.NVarChar, { length: 100, nullable: false })
        rows.forEach((row) => {
            // { code: '郵便番号', address: '住所'} の形式のデータをaddRow
            bulkLoad.addRow(row)
        })
        // バルクロード実行
        connection.execBulkLoad(bulkLoad)
    });
    return p
}
async function main() {
    // データベースに接続
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false
        }
    }
    const connection = await create_connection(config);
    const stream = fs.createReadStream('./KEN_ALL_UTF8.CSV', 'utf8')
    const reader = readline.createInterface({ input: stream })
    let rows = []
    let rowCount = 0
    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
        }
        rowCount++
    })
    reader.on('close', async () => {
        // 登録実行
        if (rows.length) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
        }
        console.log(rowCount)
        connection.close()
    })
}
main()

実行すると、半分程度しか登録されません。



読み込んだ件数12万件に対し、インサートログは60,874件。
実際にデータベースへ登録されていたのは、60,876件でした。
なんでだ？

pause / resume

BulkLoadを実行中もファイル読み込みが実行されるのでデータ件数が合わないのでは？

ドキュメントを見てみると、streamにはpauseとresumeというメソッドがあります。
https://nodejs.org/api/stream.html#stream_readable_pause

BulkLoad前にpauseを呼び出し。
終わったらresumeを呼び出して処理再開としてみます。

修正箇所の抜粋です。

    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            reader.pause() // 読み込みを一旦停止
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
            reader.resume() // 読み込み再開
        }
        rowCount++
    })

読み込んだレコード数と登録件数がかなり近づきましたが、まだ漏れがあります。

await pause

pauseが実行されるまでに読み込んでしまうデータがあるのでは？と思い、pauseにawaitをつけてみました。

    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            await reader.pause() // 読み込みを一旦停止
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
            reader.resume() // 読み込み再開
        }
        rowCount++
    })

これで読み込んだ件数と登録件数が一致してくれました。
・・・しかし、指定した件数でのデータ登録とはならず、分割境界値付近での挙動が気になります。
エディタにもawaitの意味がないという警告が表示されますし。

pause event

pauseしたときに発生するイベント内でデータ登録を行うよう修正しました。

const fs = require('fs');
const readline = require('readline')
const { Connection, TYPES } = require('tedious')
// 設定に従いデータベースへ接続
function create_connection(config) {
    
    const connection = new Connection(config)
    // Promiseをnewした時点で引数のfunctionが実行される
    const p = new Promise(function(resolve, reject) {
        connection.on('connect', err => {
            if (err) {
                reject(err)
            } else {
                resolve(connection)
            }
        });
        connection.connect()
    });
    return p
    
}
// BulkLoadの実行
function execBulkLoad(connection, rows) {
    const p = new Promise(function(resolve, reject) {
        //　BulkLoad用の設定
        const options = {}
        const bulkLoad = connection.newBulkLoad('postal_code', options, function (error, rowCount) {
            resolve('inserted ' + rowCount + ' rows (input:' + rows.length+')');
        })
        bulkLoad.addColumn('code', TYPES.Char, { nullable: false })
        bulkLoad.addColumn('address', TYPES.NVarChar, { length: 100, nullable: false })
        rows.forEach((row) => {
            // { code: '郵便番号', address: '住所'} の形式のデータをaddRow
            bulkLoad.addRow(row)
        })
        // バルクロード実行
        connection.execBulkLoad(bulkLoad)
    });
    return p
}
async function main() {
    // データベースに接続
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false
        }
    }
    const connection = await create_connection(config);
    const stream = fs.createReadStream('./KEN_ALL_UTF8.CSV', 'utf8')
    const reader = readline.createInterface({ input: stream })
    let rows = []
    let bulkRows = []
    let rowCount = 0
    reader.on('line', (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        
        if (rows.length == 10000) {
            bulkRows.push(rows)
            rows = []
            reader.pause() // 読み込みを一旦停止
        }
        rowCount++
    })
    reader.on('pause', async () => {
        // reader.pauseに加え、closeイベントの前にも呼び出される
        // 登録内容が存在しない場合はスキップ
        if (bulkRows.length == 0) {
            return
        } 
        const msg = await execBulkLoad(connection, bulkRows.pop())
        console.log(msg)
        reader.resume() // 読み込み再開
    })
    reader.on('close', async () => {
        // 登録実行
        if (rows.length > 0) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
        }
        console.log(rowCount)
        connection.close()
    })
}
main()

狙い通りの実行結果です。

fastify データベースを検索した結果をhtmlで表示する

fastifyでデータベースの検索やhtmlの表示を試してみました。
fastifyで郵便番号検索APIのサンプル
fastifyでhtmlビューを表示する(point-of-view, ejs)

2つを組み合わせて、データベースを検索した結果をhtmlで表示してみます。

サンプル

サーバー部分は以下のようになりました。

・server.js

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
fastify.register(require('point-of-view'), {
    engine: {
        ejs: require('ejs')
    }
})
fastify.get('/', async (req, reply) => {
    // 検索結果をviewに渡す
    const rows = await postal.search('銀座')
    reply.view('/views/index.ejs', { rows: rows })
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

データベースを検索する処理です。

・module/postal.js

const { Connection, Request } = require('tedious')
module.exports = () => {
    // 接続情報
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false,
            rowCollectionOnRequestCompletion : true
        }
    }
    // 設定に従いデータベースへ接続
    this.create_connection = (config) => {
    
        const connection = new Connection(config)
        // Promiseをnewした時点で引数のfunctionが実行される
        const p = new Promise(function(resolve, reject) {
            connection.on('connect', err => {
                if (err) {
                    reject(err)
                } else {
                    resolve(connection)
                }
            })
            connection.connect()
        })
        return p
        
    }
    // SQLの実行
    this.execute = (connection, sql) => {
        const p = new Promise(function(resolve, reject) {
            const request = new Request(sql, (err, rowCount, columns) => {
                if (err) {
                    reject(err)
                    return
                }
                let rows = []
                columns.forEach(column => {
                    let row = {}
                    column.forEach(field => {
                        row[field.metadata.colName] = field.value
                    });
                    rows.push(row)
                });
                
                resolve(rows)
            })
            connection.execSql(request)
        });
        return p
    }
    // 住所の一部でデータを検索
    this.search = async (address) => {
        const connection = await this.create_connection(config)
        const rows = await this.execute(connection, "SELECT * FROM postal_code WHERE address LIKE '%" + address + "%' ORDER BY code")
        connection.close()
        return rows
    }
    return this
}

表示部分はこのようになりました。

・views/index.ejs

<html lang="ja">
<head>
<meta charset="UTF-8">
<title>fastifyサンプル</title>
<style>
table {
    border: solid 2px;
    border-collapse: collapse;
}
th, td {
    border: solid 2px;
    padding: 4px;
}
</style>
</head>
<body>
    <table>
        <thead>
            <tr>
                <th>郵便番号</th>
                <th>住所</th>
            </tr>
        </thead>
        <tbody>
            <% rows.forEach(row => { %>
                <tr>
                    <td><%= row.code %></td>
                    <td><%= row.address %></td>
                </tr>
            <% }) %>
        </tbody>
    </table>
</body>
</html>

サーバーを起動してブラウザで表示してみると、ちゃんと住所に「銀座」を含む結果が表示されました。

fastify 入力データの検証(validation)

fastifyで入力データの検証を試してみます。

id:数値
address:文字列 必須

というパターンだとこんな感じになりました。
受け取ったデータをそのまま送り返しています。

const fastify = require('fastify')({ logger: true })
const opts = {
    schema: {
        body: {
            type: 'object',
            required: [ // addressを必須項目に
                'address'
            ],
            properties: {
                id: { type: 'number'},
                address: { type: 'string' }
            }
        }
    }
}
fastify.post('/', opts, async (req, reply) => {
    // jsonデータはreq.bodyに設定される
    return req.body
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

正常系


idが文字列(数値へ変換可能)

自動的に数値へ変換されました。

idが文字列(数値へ変換不可)

ちゃんとエラーになりました。

idがなし


addressが空白文字列


addressが数値


addressがなし



このvalidationは、「Ajv」を使用しているとのことで、詳しい使い方はこちらのドキュメントを見たほうが良さそうです。
https://ajv.js.org/json-schema.html

fastify json形式のPOSTデータ受け取り

fastifyはPOSTデータはjson形式を想定しており、x-www-form-urlencoded形式のデータ受信にはひと工夫必要でした。
fastifyでPOSTデータ受信時、FST_ERR_CTP_INVALID_MEDIA_TYPE

POSTデータがjson形式の場合について見てみます。

プログラム修正

{"address":"銀座"}

という形式で検索パラメーターを送信することにします。
json形式のPOSTデータは自動的にパースされ、req.bodyに設定されました。

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
fastify.post('/', async (req, reply) => {
    // jsonデータはreq.bodyに設定される
    const address = req.body.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

動作確認

fastifyでPOSTデータ受信時、FST_ERR_CTP_INVALID_MEDIA_TYPE

fastifyで郵便番号検索APIを作ってみました。
fastifyで郵便番号検索APIのサンプル

検索する住所をGETのクエリーパラメーターではなく、POSTで送信するよう変更してみます。

最初の実装

「fastify.getをfastify.postに変えれば動くだろう」と思い変更してみました。

・server.js

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
/*
fastify.get('/', async (req, reply) => {
    // クエリーパラメーター取得
    const address = req.query.address
    const rows = await postal.search(address)
    return rows
})
*/
// postに変更
fastify.post('/', async (req, reply) => {
    const address = req.query.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

curlで動作を確認すると、FST_ERR_CTP_INVALID_MEDIA_TYPE
Unsupported Media Type: application/x-www-form-urlencoded
というエラーになります。

POST対応

調べてみると、デフォルトでPOSTはjson形式のみを受け付け、x-www-form-urlencodedなデータはエラーになるようです。
multipartも同様の模様。
Giving error with node/fastify Unsupported Media Type: application/x-www-form-urlencoded

fastify-formbodyを使うといいよとのこと。
fastify-formbody

インストールします。




インストールしたfastify-formbodyを登録し、POSTデータをreq.bodyから取得するよう修正します。

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
// POST対応
fastify.register(require('fastify-formbody'))
// postに変更
fastify.post('/', async (req, reply) => {
    // req.bodyからPOSTデータを取得するよう変更
    //const address = req.query.address
    const address = req.body.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

これでPOSTデータを受け取り、検索が実行できるようになりました。