S3のcsvをストリーム取り込みする際1行目を読み飛ばして、変換の後、転送したいが読み飛ばせない問題

Question

前提

S3にアップロードされたcsvを取り込む際
1行目を飛ばし2行目から読み込み
変換処理をかけようとした際に
エラーが発生した

実現したいこと

ここに実現したいことを箇条書きで書いてください。

• S3にアップロードされたcsvを読み込み
• 1行目（項目、ヘッダー）を飛ばし、2行目から読み込み
• 項目に応じた変換処理をかけOpensearchへ転送したい

発生している問題・エラーメッセージ

本Lamda(python)はトリガーを設定しています。

【トリガー】
S3にcsvがアップロードされることで、Lamda(python)が実行開始される

アップロードするcsv

vmanage_host, systemip, hostname , devicemodel, bfdSessionsUp, bfdSessions, datetime, IPsecTunnel
vmanageABC, 123.456.789.012, vedgeABC, deviceABC, 1, 2, 3, 2022/1/1 01:01:01, 123

エラー（本題）
１行目が項目のため、型変換できないというエラーとなっている（と推察）

[Error] ValueError : invalid literal for int() with base 10: `ABC.csv`

該当のソースコード

import boto3
import re
import requests
import csv
import io
import BytesIO
from requests_aws4auth import AWS4Auth

#配列
vmanage_host = []
systemip = []
hostname = []
devicemodel = []
bfdSessionsUp = []
bfdSessions = []
datetime = []
IPsecTunnel = []


region = '（略）' # e.g. us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

host = '（略）' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-s3-index'
type = '_doc'
url = host + '/' + index + '/' + type

headers = { "Content-Type": "application/json" }

s3 = boto3.client('s3')

# Lambda execution starts here
def handler(event, context):
    for record in event['Records']:
        #header = next(reader)  ←検証③

        # Get the bucket name and key for the new file
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Get, read, and split the file into lines
        obj = s3.get_object(Bucket=bucket, Key=key)
        body = obj['Body'].read()
        textIo = io.TextIOWrapper(io.BytesIO(body))
        #header = next(reader)  ←検証①

        # Match the regular expressions to each line and index the JSON
        for row in lines csv.reader(textIo):
            #header = next(reader)  ←検証②

            print(row)
            vmanage_host = row[0]
            systemip = row[1]
            hostname = row[2]
            devicemodel = row[3]
            bfdSessionsUp = row[4]
            bfdSessions = row[5]
            datetime = row[6]
            IPsecTunnel = row[7]
            
            print(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]) 

            document = {
                         "vmanage_host ": vmanage_host ,
                         "systemip  ": systemip ,
                         "hostname ": hostname ,
                         "devicemodel ": devicemodel ,
                         "bfdSessionsUp": [int(bfdSessionsUp)],
                         "bfdSessions ": [int(bfdSessions)],
                         "datetime ": [strftime(datetime)],
                         "IPsecTunnel ": [int(IPsecTunnel)],
                       }
            r = requests.post(url, auth=awsauth, json=document, headers=headers)

試したこと

検証箇所①、②、③にそれぞれ以下のヘッダーを読み飛ばす処理を
いれて、確認したが、いずれの箇所も同様のメッセージとなり実行を
確認することができなかった。

header = next(reader)

補足情報（FW/ツールのバージョンなど）

Python3.8

Comments

[melian]

エラーが発生しているのは以下で、

for record in event['Records']:

event に Records というキーがない、と表示されています。print(event.keys()) などとして、内容を確認してみてはどうでしょうか。

[msy47]

melianさん
コメントありがとうございます。
確認が遅れてすみません。現在確認中です。

[msy47]

melianさん
ご助言ありがとうございます。再度ご教授を頂きたく。
以下のそれぞれの箇所にいれたのですが、エラー内容は変わらず。
Print結果も表示されませんでした。挿入箇所はまちがいないでしょうか。

def handler(event, context):
for record in event['Records']:
print(event.keys()) ←挿入①

# Get the bucket name and key for the new file
bucket = record['s3']['bucket']['name']
print(event.keys()) ←挿入②

[melian]

for 文の直前に入れてみて下さい。

def handler(event, context):
print(event.keys()) # ここに挿入
for record in event['Records']:

[msy47]

melianさん早いご回答ありがとうございます。
結果は以下となりました。Printつけてもつけなくても同様のエラー内容となっております。

Response
{
"errorMessage" : "`Records`",
"errorType" : "KeyError",
"requestId" : "(略)"
"stackTrace" : [
" File\" /var/task/sample.py\, line 39, inhander\n for record in event[`Records`]:\n"
]
}

[melian]

AWS Lambda には詳しくないのでよく分かっていなのですが、print(event.keys()) の実行結果は AWS Cloud Watch の方に出力されているのではないでしょうか。

[msy47]

melianさん
ご指摘ありがとうございます。
お恥ずかしい限りです。CloudWatchに出力されておりました。

dict_keys([`key1`, `key2`, `key3`])

[melian]

そうなりますと、やはり event には 'Records' というキーが存在しないということになりますね。。。

[msy47]

元ソースはこちらをベースにして改良してきました。
Amazon OpenSearch Service へのストリーミングデータをロードする - Amazon OpenSearch Service
https://docs.aws.amazon.com/ja_jp/opensearch-service/latest/developerguide/integrations.html#integrations-s3-lambda

＞event には 'Records' というキーが存在しないということ
すみません。この意図するところ、重要性が理解できておらず。。。

[melian]

今回の問題は、Records というキーが存在しないために以下の部分(event['Records'])で必ずエラーが発生してしまうことなのです。

for record in event['Records']:

その Amazon OpenSearch Service の example code では確かに Records キーを参照しているので、、とここまで書いて気がついたのですが、example code の次に、

> region と host の変数を編集します。

と書かれています。こちらは編集済みでしょうか？

[msy47]

melianさん

> region と host の変数を編集します。
はい、こちら設定済みとなります。
セキュリティのため未記載としております。

重大なことを記載しておりませんでした。
こちらトリガーを設定しておりました。

トリガー
S3の csv/に「S3にアップロードされているcsv」がアップロードされると、本Lambdaが実行されます。

別途エラーが出てきたため整理致します。

[msy47]

melianさん

本来のエラー（型変換できない）
→回避するためのロジックの模索
を修正・追記いたしました。

エラー内容型変換できない
→１行目（項目）のため１行目を飛ばし、２行目から転送するロジックが必要
→header = next(reader)　はcsvのみ？ストリーム形式では使用できない？
[Error] ValueError : invalid literal for int() with base 10: `ABC.csv`

Answer 1

以下の箇所に
skiptext = next(textIo)
を入れることで解消することができました。
正解かどうかは不明・・・

    textIo = io.TextIOWrapper(io.BytesIO(body))

    skiptext = next(textIo)

    # Match the regular expressions to each line and index the JSON
    for row in lines csv.reader(textIo):