LoginSignup
5
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@Sicut_study(渡邉 臣 | JISOU)

AWS初心者による実践AWSデータサイエンスのつまずいたメモ

Last updated at Posted at 2021-12-02

はじめに

この記事ではAWS初心者の私が「実践AWSデータサイエンス」の本を読んでつまずいたところをメモしていく記事となります。
対象は書籍10章までとなります(これから先は利用予定が今のところないため割愛)

こちらの書籍内容は求めていたものですが、省略が多く(実践なので)、AWSの基本的な使い方については書かれていないように思えます。そこで試行錯誤しながらゆっくりメモしていければと思います。

準備

以下のGitにある01_introduction/のコードをすべてSageMaker studioで実行する必要があります。
するとロールやバケットなどをいい感じに作成してくれます。

私はすべて1から画面上で作成していますのでそのような方向けのエラーも含まれています。

3章 SageMaker Autopilot

データセットの取得について

SageMaker Studioを起動して、以下のファイルを開く
https://github.com/data-science-on-aws/oreilly_book/blob/master/03_automl/01_Prepare_Dataset_Autopilot.ipynb

そのあとすべて実行することで作成できる(前半はエラーになるが、途中からダウンロードできる)

PythonSDKでSageMakerAutopilotをしたときにS3にアクセスできない

role/service-role/awsdeepracersagemakeraccessrole has "s3:putobject" permissions on the bucket.

というエラーが発生。S3とSagemakerのフルアクセスをIAMに設定しているのにダメだった。
【機械学習初級】Amazon SageMaker導入の手順~S3にデータをアップロード編~

こちらの記事を発見して以下の内容を実施したところエラーが解決した(ただしSでなくsでないとエラーになる)

S3バケットを作成するうえで、注意する点が1点あります。
それは、バケット名に「Sagemaker」を含めることです。
こうすることでノートブックインスタンスを作成する際に設定をしたデフォルトで設定されているIAMポリシーでSagemakerがS3バケットにアクセスする権限を付与することができます。

Athenaでmismatched input 'FUNCTION'. Expecting: 'EXTERNAL'

Athenaのテーブル名に-を使っていたことが原因だった

aws-handson → aws_hands

に変更して対応した

Athenaで結果が空白や文字列が数値になって返ってくる

これは入力のS3とクエリの履歴保存先であるS3が同じになっているのが原因
入力のS3はファイルのみしか置いてはいけない

エディタの設定から出力先のS3を設定して解決

AthenaでSAGEMAKER_INVOKE_ENDPOINTが呼び出せない

ここはかなり時間がかかりました。以下の点に注意して行うとできます。

  1. リージョンはオレゴンで行う
    対応しているリージョンが決まっていました

  1. データセットは記号を置換しておく
    区切り文字として扱われてしまうことがあり途中でエラーの原因になっていました
df['review_body'] = df['review_body'].apply(lambda x: re.sub(r"[^a-zA-Z0-9]"," ",x))
  1. クエリを変更する
    そもそもクエリが新しくなっているようです。

USING EXTERNAL FUNCTION predict_rating(review_body VARCHAR) 
    RETURNS INT
    SAGEMAKER 'product-reviews-sample'
SELECT *, predict_rating(REPLACE(review_body, ',', ' ')) AS pred
FROM "cheat_detection_sample"."product_reveiws" limit 100;

また、デバックのために以下のハンズオンを参考にして動くことを確認しました。

Athenaのテーブルやデータベースが消せない

出力のS3の設定をしていなかった

Redshifのデータ入力までの手順

初期設定は以下を参考にしました。
[AWS]はじめてのAmazon Redshift（アマゾン・レッドシフト）

CSVの取り込みは以下の記事を参考にしました。
RedshiftでCOPYコマンドを試してみた

そして、かなり沼ったのがデータについてです。Athenaでamazon_reviews_us_Digital_Software_v1_00_autopilot.csvを使いましたが、このまま利用するとジョブ処理でfailedしてしまいます。

そこで、記号をすべて置換したり、文字数を制限したりとデバックをして5文字までならエラーにならないことを突き止めました。ここからは予測ですが以下が原因かと思います。

Autopilotは自然言語処理の単語をベクトルにする処理が実装されておらず、文字列があった場合にOneHotエンコーディングをしているのではないかと思います。5文字までなら同じ行があるので意味のあるベクトルになりますが、それ以上だとすべてが違うため判断ができず、パイプラインが作成できないというエラーが出るかと思います。(すべて同じ文字列でも確かめましたが、これはこれで同じですみたいなエラーがでました)

⇒Amazon Comprehendを使ってならAutopilotで自然言語が扱えますが、この方法では無理でした

そこで、amazon_reviews_us_Digital_Software_v1_00_autopilot.csvのすべての項目でCSVを作成し、入力したところエラーがでなくなりました。

ちなみにSQLは以下を入力しました。

# テーブル作成
create table product_reviews(
  marketplace VARCHAR(128),
  customer_id VARCHAR(128),
  review_id VARCHAR(128),
  product_id VARCHAR(128),
  product_parent VARCHAR(128),
  product_title VARCHAR(128),
  product_category VARCHAR(128),
  star_rating INT,
  helpful_votes INT,
  total_votes INT,
  vine VARCHAR(128),
  verified_purchase VARCHAR(128),
  review_headline VARCHAR(10000),
  review_body VARCHAR(60000),
  review_date VARCHAR(128)
) 

# CSV取り込み
copy product_reviews from 's3://path' CSV
iam_role 'arn:aws:iam::(ロール)'
region 'us-west-2' # オレゴンでやりましたがオハイオでもOK
IGNOREHEADER 1;

# 学習
CREATE MODEL モデル名
FROM product_reviews
TARGET star_rating
FUNCTION 関数名
IAM_ROLE 'arn:aws:iam::(ロール)'
SETTINGS (
  S3_BUCKET 'S3バケット名'
);

# 予測 (10行だけ予測)
SELECT 関数名(marketplace, customer_id, review_id, product_id, product_parent, product_title, product_category, helpful_votes, total_votes, vine, verified_purchase, review_headline, review_body, review_date)
FROM product_reviews limit 10

4章 クラウドへのデータの取り込み

AthenaのParaquetベースのテーブル生成ができない

Gitに載っていたほうで動かした

CREATE TABLE IF NOT EXISTS dsoaws.amazon_reviews_parquet_from_tsv
WITH (format = 'PARQUET', external_location = 's3://sagemaker-aws/amazon-reviews-pds/parquet-from-tsv', partitioned_by = ARRAY['product_category']) AS
SELECT marketplace,
         customer_id,
         review_id,
         product_id,
         product_parent,
         product_title,
         star_rating,
         helpful_votes,
         total_votes,
         vine,
         verified_purchase,
         review_headline,
         review_body,
         CAST(YEAR(DATE(review_date)) AS INTEGER) AS year,
         DATE_DIFF('day', DATE('1970-01-01'), DATE(review_date)) AS review_date,
         product_category
FROM amazon_reviews_tsv

Glue Crawlerで権限エラー

まずやったのはSageMakerからJupyterを開いてクローラー作成のコード行しました。

AccessDeniedException: An error occurred (AccessDeniedException) when calling the CreateCrawler operation: User: arn:aws:sts::*******:assumed-role/AWSDeepRacerSageMakerAccessRole/SageMaker is not authorized to perform: iam:PassRole on resource: arn:aws:iam::*******:role/service-role/AWSDeepRacerSageMakerAccessRole because no identity-based policy allows the iam:PassRole action

アクセス権限が内容でした。JupyterのIAMにAmazonS3FullAccess, AmazonAthenaFullAccess, AWSGlueConsoleFullAccessなど加えてもできず、利用しているIAMにも設定しましたがダメでした。

次にローカルのPythonでboto3をインストールして実行しました。

botocore.errorfactory.InvalidInputException: An error occurred (InvalidInputException) when calling the CreateCrawler operation: Service is unable to assume role arn:aws:iam::****:user/watanabe.jin. Please verify role's TrustPolicy

やはりエラーになりました。roleの設定がうまくいっていないようでした。

role = sagemaker.get_execution_role()

これも動かず。

調べていたところ以下の記事を発見。新たなGlueのロールを作成して設定しました。
ステップ 2: AWS Glue 用にIAM ロールを作成する

そしてローカルで作成したIAMをroleに設定したところ成功。(Jupyterはだめでした)

5章 データセットを探索

RedshiftとSagemakerが接続できない (調査中)

pd.read_sql_queryを行うと、以下のようなエラーが発生しました。

OperationalError: (psycopg2.OperationalError) could not connect to server: Connection timed out
	Is the server running on host "dsoaws.c1q5xyrhck7t.us-west-2.redshift.amazonaws.com" (172.31.0.190) and accepting
	TCP/IP connections on port 5439?

(Background on this error at: http://sqlalche.me/e/13/e3q8

セキュリティグループでport5439を開けたり、同じVPCに所属させたりと色々試したのですがダメでした。

こちらの01_introductionのコードを全て実行して、04_ingestの07_Load_TSV_Data_From_Athena_Into_Redshift.ipynbを実行したところ同じエラーが発生しました。
技術監修の方が親切にも連絡をくれまして、動いていただけそうでした。

おそらくAWS側の問題な気がしますがこちらは進捗があり次第更新します。
Redshiftを利用しているところは書籍の一部なので先に進めるとのことでした。

以下が同じ問題で作成されていたIssueになります。

● 12/8 更新
原因と回避方法を見つけていただき対応ができました。
上のIssueに詳しくは書いてありますが、ここでは細かい修正箇所を載せます。

まず、04_ingest/07_Load_TSV_Data_From_Athena_Into_Redshift.ipynb

# はじめに必要なライブラリのインストール
!pip install awswrangler --upgrade  # default version does not support data api

# 1つめ
s.execute(statement)
s.commit()
↓
data_client = boto3.client('redshift-data')
result = data_client.execute_statement(               # <----- note this is an asynchronous api
    ClusterIdentifier=redshift_cluster_identifier,
    Database=database_name_redshift,
    DbUser=master_user_name,
    Sql=statement,
)

# 2つめ
df = pd.read_sql_query(statement, engine)
df.head(5)
↓
import awswrangler as wr

con_redshift = wr.data_api.redshift.connect(
    cluster_id=redshift_cluster_identifier,
    database=database_name_redshift,
    db_user=master_user_name,
)

df = wr.data_api.redshift.read_sql_query(
    sql=statement,
    con=con_redshift,
)

# 3つめ
# Create table function, pass session, table name prefix and start & end year

def create_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
        CREATE TABLE IF NOT EXISTS redshift.{}( 
             marketplace varchar(2),
             customer_id varchar(8),
             review_id varchar(14),
             product_id varchar(10) DISTKEY,
             product_parent varchar(9),
             product_title varchar(400),
             product_category varchar(24),
             star_rating int,
             helpful_votes int,
             total_votes int,
             vine varchar(1),
             verified_purchase varchar(1),
             review_headline varchar(128),
             review_body varchar(65535),
             review_date varchar(10),
             year int)  SORTKEY (product_category)
        """.format(current_table_name)

        #print(statement)
        session.execute(statement)
    session.commit()
        
    print("Done.")
↓
def create_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
        CREATE TABLE IF NOT EXISTS redshift.{}( 
             marketplace varchar(2),
             customer_id varchar(8),
             review_id varchar(14),
             product_id varchar(10) DISTKEY,
             product_parent varchar(9),
             product_title varchar(400),
             product_category varchar(24),
             star_rating int,
             helpful_votes int,
             total_votes int,
             vine varchar(1),
             verified_purchase varchar(1),
             review_headline varchar(128),
             review_body varchar(65535),
             review_date varchar(10),
             year int)  SORTKEY (product_category)
        """.format(current_table_name)

        #print(statement)
#         session.execute(statement)
        result = data_client.execute_statement(               # <----- note this is an asynchronous api
        ClusterIdentifier=redshift_cluster_identifier,
        Database=database_name_redshift,
        DbUser=master_user_name,
        Sql=statement,
        )
        
    print("Done.")

# 4つめ
# INSERT INTO function, pass session, table name prefix and start & end year

def insert_into_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        print(year)
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
            INSERT 
            INTO
                redshift.{}
                SELECT
                    marketplace,
                    customer_id,
                    review_id,
                    product_id,
                    product_parent,
                    product_title,
                    product_category,
                    star_rating,
                    helpful_votes,
                    total_votes,
                    vine,
                    verified_purchase,
                    review_headline,
                    review_body,
                    review_date,
                    CAST(DATE_PART_YEAR(TO_DATE(review_date, 'YYYY-MM-DD')) AS INTEGER) AS year
                FROM
                    athena.amazon_reviews_tsv             
                WHERE
                    year = {}
            """.format(current_table_name, year)
        #print(statement)
        session.execute(statement)
        session.commit()        
    print("Done.")
↓
# INSERT INTO function, pass session, table name prefix and start & end year

def insert_into_redshift_table_tsv(session, table_name_prefix, start_year, end_year):
    for year in range(start_year, end_year + 1, 1):
        print(year)
        current_table_name = table_name_prefix+'_'+str(year)
        statement = """
            INSERT 
            INTO
                redshift.{}
                SELECT
                    marketplace,
                    customer_id,
                    review_id,
                    product_id,
                    product_parent,
                    product_title,
                    product_category,
                    star_rating,
                    helpful_votes,
                    total_votes,
                    vine,
                    verified_purchase,
                    review_headline,
                    review_body,
                    review_date,
                    CAST(DATE_PART_YEAR(TO_DATE(review_date, 'YYYY-MM-DD')) AS INTEGER) AS year
                FROM
                    athena.amazon_reviews_tsv             
                WHERE
                    year = {}
            """.format(current_table_name, year)
        #print(statement)
#         session.execute(statement)
#         session.commit()        
        result = data_client.execute_statement(               # <----- note this is an asynchronous api
            ClusterIdentifier=redshift_cluster_identifier,
            Database=database_name_redshift,
            DbUser=master_user_name,
            Sql=statement,
        )
    print("Done.")

次にこのあと使うコードも修正があるので載せておきます。(05_exprole/redshift/02_Explore_Redshift_Data.ipynb)

# 準備
!pip install awswrangler --upgrade

# 1つめ
%%time
df = pd.read_sql_query(
    """SELECT approximate count(distinct customer_id)
                                FROM {}.{}
                                GROUP BY product_category""".format(
        redshift_schema, redshift_table_2015
    ),
    engine,
)
↓
import awswrangler as wr
statement = """SELECT approximate count(distinct customer_id)
                                FROM {}.{}
                                GROUP BY product_category""".format(redshift_schema, redshift_table_2015)

con_redshift = wr.data_api.redshift.connect(
    cluster_id=redshift_cluster_identifier,
    database=redshift_database,
    db_user=redshift_host,
)

df = wr.data_api.redshift.read_sql_query(
    sql=statement,
    con=con_redshift,
)

# 2つめ
%%time
df = pd.read_sql_query(
    """SELECT count(distinct customer_id)
                                FROM {}.{}
                                GROUP BY product_category""".format(
        redshift_schema, redshift_table_2015
    ),
    engine,
)

↓
%%time

statement =     """SELECT count(distinct customer_id)
                                FROM {}.{}
                                GROUP BY product_category""".format(redshift_schema, redshift_table_2015)

con_redshift = wr.data_api.redshift.connect(
    cluster_id=redshift_cluster_identifier,
    database=redshift_database,
    db_user=redshift_host,
)

df = wr.data_api.redshift.read_sql_query(
    sql=statement,
    con=con_redshift,
)

# 3つめ
df = pd.read_sql_query(statement, engine)
↓
con_redshift = wr.data_api.redshift.connect(
    cluster_id=redshift_cluster_identifier,
    database=redshift_database,
    db_user=redshift_host,
)

df = wr.data_api.redshift.read_sql_query(
    sql=statement,
    con=con_redshift,
)

# 4つめ
df = pd.read_sql_query(statement, engine)
df.head(5)
↓
con_redshift = wr.data_api.redshift.connect(
    cluster_id=redshift_cluster_identifier,
    database=redshift_database,
    db_user=redshift_host,
)

df = wr.data_api.redshift.read_sql_query(
    sql=statement,
    con=con_redshift,
)
df.head(5)

05_exploler/redshift/04_Visualize_Reviews_Dataset_Redshift.ipynbも同じ要領でやってください

Amazon QuickSightで結果が表示できない

QuickSightの設定は以下に書かれている

始めはグラフが表示されずエラーの詳細をみるとS3へのアクセス権限がないと言われたので、作成したIAMにS3FullAccessを付与して解決

始めはトライアルだが、お金がかかるのでアカウント削除と関連したロール、ポリシーの削除を忘れずに

Deequが動かない

05_explore/sparkに05_explore/archive/にある設定ファイルを移動することでdeequコンテナをたてることができます。

そのあとに制約を指定するコードはpreprocess-deequ-pyspark.pyの内容でした

Glue DataBrewの動かし方

ここに載っています。
https://github.com/data-science-on-aws/oreilly_book/blob/master/05_explore/data-brew/05_Analyze_Data_Glue_DataBrew.ipynb

  • 北部リージョンを利用する (データセット作成時のs3が北部になるため)
  • 利用しているロールにAwsGlueDataBrewFullAccessPolicyを追加
  • IAMの信頼関係から信頼関係を編集して以下を追加
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "databrew.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }

6章 モデル訓練のためのデータセット準備

ここからは日本語版のquickハンズオンも対応していますので、適宜参考にするとよいかと思います。

6.6のコードがみつからない

公式のGitの/05_exproler/data-brewにコードが載っています

7章 最初のモデル訓練

JumpStartのやり方

SageMakerのJumpStartからTextModelsにあるBERT Base Uncased(classification)を選択する
そのあと、Trainというがあるので、そこからS3のバケットを選択する
S3のバケットは適当に作成して、こちらのデータを追加する
そしてAutopilotで学習を行う(30分くらい)

終わったら、デプロイをしてエンドポイントの名前を取得
oreilly_book/07_train/jumpstart/99_Predict_Jumpstart.ipynbのエンドポイントを変更することで可能

9章 モデルを本番環境にデプロイ

09_deploy/01_Invoke_SageMaker_Autopilot_Model_From_AthenaのSQL文

SQLを以下のように修正

statement = """
USING EXTERNAL FUNCTION predict_star_rating(review_body VARCHAR) 
    RETURNS VARCHAR
    SAGEMAKER '{}'

SELECT review_id, review_body, predict_star_rating(REPLACE(review_body, ',', ' ')) AS predicted_star_rating 
    FROM {}.{} LIMIT 10
    """.format(
    autopilot_endpoint_name, database_name, table_name
)

print(statement)

10章 パイプラインとMLOps

03_Detect_Model_Bias_Clarifybias.ipynbでバイアスのデータがみつからない

test_data_bias_path = "./data-clarify/test_data_bias.jsonl"

このデータがみつからないのでエラーになった。07_train/data-clarifyをコピーして持ってきたがProcessingJobで以下のエラー

UnexpectedStatusException: Error for Processing job Clarify-Posttraining-Bias-2021-12-22-06-56-21-917: Failed. Reason: ClientError: facet name_or_index:'product_category' cannot be the same as group_variable. Please specify a facet other than group_variable. group_variable: product_category

おそらく、product_category以外のカラムも含めて出力するように7章で調節しないといけない
04_Explain_Model_SHAP_Clarify.ipynbも同様だがこちらは動きはするがやはり結果から考えてそうかと思われる。

おわりに

SageMakerで何がどのような流れでできるか学べる本でした。
苦戦はしましたが、やり方さえ分かっていればRedshift APIのところ以外は問題ないかなという感じでした。
やれることがわかったら次は自分の環境で同じ手順を踏んでいく作業が始まるので頑張りたいと思います。

参考

5
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?