# ===== 環境変数の設定 ===== # Linux/Mac export AWS_ACCESS_KEY_ID=”your_access_key” export AWS_SECRET_ACCESS_KEY=”your_secret_key” export AWS_DEFAULT_REGION=”ap-northeast-1″ # Windows（コマンドプロンプト） set AWS_ACCESS_KEY_ID=your_access_key set AWS_SECRET_ACCESS_KEY=your_secret_key set AWS_DEFAULT_REGION=ap-northeast-1 # Windows（PowerShell） $env:AWS_ACCESS_KEY_ID=”your_access_key” $env:AWS_SECRET_ACCESS_KEY=”your_secret_key” $env:AWS_DEFAULT_REGION=”ap-northeast-1″

# ===== S3への基本的なアップロード ===== import boto3 import pandas as pd # サンプルデータ df = pd.DataFrame({ ‘商品ID’: [1, 2, 3, 4, 5], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’, ‘いちご’], ‘価格’: [150, 100, 80, 300, 250], ‘在庫’: [50, 100, 80, 30, 40] }) print(“=== アップロードするデータ ===”) print(df) # まずローカルにCSVファイルを作成 df.to_csv(‘products.csv’, index=False, encoding=’utf-8-sig’) # S3クライアントを作成（環境変数から自動で認証情報を取得） s3 = boto3.client(‘s3’) # S3にアップロード bucket_name = ‘my-data-bucket’ # あなたのバケット名 file_name = ‘products.csv’ # ローカルファイル名 s3_path = ‘data/products.csv’ # S3内のパス s3.upload_file(file_name, bucket_name, s3_path) print(f”\n✅ S3にアップロードしました！”) print(f” パス: s3://{bucket_name}/{s3_path}”)

# ===== メモリから直接S3にアップロード ===== import boto3 import io import pandas as pd # サンプルデータ df = pd.DataFrame({ ‘商品ID’: [1, 2, 3], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘価格’: [150, 100, 80] }) # DataFrameをCSV文字列に変換 csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) # S3にアップロード s3 = boto3.client(‘s3′) s3.put_object( Bucket=’my-data-bucket’, Key=’data/products_direct.csv’, Body=csv_buffer.getvalue().encode(‘utf-8’) ) print(“✅ メモリから直接S3にアップロードしました！”)

# ===== S3からファイルをダウンロード ===== import boto3 import pandas as pd s3 = boto3.client(‘s3’) # 方法1：ファイルに保存してから読み込み s3.download_file( ‘my-data-bucket’, # バケット名 ‘data/products.csv’, # S3内のパス ‘downloaded_products.csv’ # ローカルに保存するファイル名 ) df = pd.read_csv(‘downloaded_products.csv’) print(“=== ダウンロードしたデータ ===”) print(df)

# ===== S3内のファイル一覧を取得 ===== import boto3 s3 = boto3.client(‘s3′) response = s3.list_objects_v2( Bucket=’my-data-bucket’, Prefix=’data/’ # このプレフィックスで始まるファイルのみ ) if ‘Contents’ in response: print(“=== S3内のファイル一覧 ===”) for obj in response[‘Contents’]: size_kb = obj[‘Size’] / 1024 print(f” {obj[‘Key’]} ({size_kb:.1f} KB)”) else: print(“ファイルが見つかりませんでした”)

# 環境変数で認証情報を設定 export GOOGLE_APPLICATION_CREDENTIALS=”/path/to/your-service-account-key.json”

# ===== GCSへの基本的なアップロード ===== from google.cloud import storage import pandas as pd # サンプルデータ df = pd.DataFrame({ ‘商品ID’: [1, 2, 3], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘価格’: [150, 100, 80] }) # ローカルにCSVを保存 df.to_csv(‘products.csv’, index=False, encoding=’utf-8-sig’) # GCSクライアントを作成 client = storage.Client() # バケットを取得 bucket = client.get_bucket(‘my-gcs-bucket’) # ファイルをアップロード blob = bucket.blob(‘data/products.csv’) blob.upload_from_filename(‘products.csv’) print(f”✅ GCSにアップロードしました！”) print(f” パス: gs://my-gcs-bucket/data/products.csv”)

# ===== メモリから直接GCSにアップロード ===== from google.cloud import storage import io import pandas as pd df = pd.DataFrame({ ‘商品ID’: [1, 2, 3], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘価格’: [150, 100, 80] }) # DataFrameをCSV文字列に変換 csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) # GCSにアップロード client = storage.Client() bucket = client.get_bucket(‘my-gcs-bucket’) blob = bucket.blob(‘data/products_direct.csv’) blob.upload_from_string(csv_buffer.getvalue(), content_type=’text/csv’) print(“✅ メモリから直接GCSにアップロードしました！”)

# ===== 日付でパーティショニングしてアップロード ===== import boto3 import io import pandas as pd from datetime import datetime def upload_with_date_partition(df, bucket_name, base_path, date=None): “”” 日付ごとにパーティショニングしてS3にアップロード Parameters: ———– df : DataFrame アップロードするデータ bucket_name : str S3バケット名 base_path : str ベースパス date : datetime or None 日付（Noneの場合は今日） Returns: ——– str : S3パス “”” if date is None: date = datetime.now() # パーティションパスを作成 year = date.strftime(‘%Y’) month = date.strftime(‘%m’) day = date.strftime(‘%d’) s3_path = f”{base_path}/year={year}/month={month}/day={day}/data.csv” # CSVに変換 csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) # S3にアップロード s3 = boto3.client(‘s3’) s3.put_object( Bucket=bucket_name, Key=s3_path, Body=csv_buffer.getvalue().encode(‘utf-8’) ) print(f”✅ アップロード完了”) print(f” パス: s3://{bucket_name}/{s3_path}”) return s3_path # 使用例 df = pd.DataFrame({ ‘売上ID’: [1, 2, 3], ‘金額’: [1000, 2000, 1500] }) upload_with_date_partition(df, ‘my-data-bucket’, ‘sales’, datetime(2024, 1, 15))

# ===== 日付と地域でパーティショニング ===== def upload_with_multi_partition(df, bucket_name, base_path, date, region): “””日付と地域でパーティショニング””” year = date.strftime(‘%Y’) month = date.strftime(‘%m’) day = date.strftime(‘%d’) # パーティションパス（地域 → 年 → 月 → 日） s3_path = f”{base_path}/region={region}/year={year}/month={month}/day={day}/data.csv” # CSV変換してアップロード csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) s3 = boto3.client(‘s3’) s3.put_object( Bucket=bucket_name, Key=s3_path, Body=csv_buffer.getvalue().encode(‘utf-8’) ) print(f”✅ {region}の{date.strftime(‘%Y-%m-%d’)}データをアップロード”) return s3_path # 使用例 from datetime import datetime tokyo_df = pd.DataFrame({‘売上’: [1000, 2000]}) osaka_df = pd.DataFrame({‘売上’: [1500, 2500]}) today = datetime(2024, 1, 15) upload_with_multi_partition(tokyo_df, ‘my-data-bucket’, ‘sales’, today, ‘tokyo’) upload_with_multi_partition(osaka_df, ‘my-data-bucket’, ‘sales’, today, ‘osaka’)

# ===== 日次売上パイプライン ===== import pandas as pd import numpy as np from datetime import datetime, timedelta import boto3 import io def generate_daily_sales(date): “””指定日の売上データを生成（ダミー）””” return pd.DataFrame({ ‘日付’: [date.strftime(‘%Y-%m-%d’)] * 10, ‘商品ID’: range(1, 11), ‘売上’: np.random.randint(1000, 10000, 10) }) def aggregate_daily_sales(df): “””日次売上を集計””” return pd.DataFrame({ ‘日付’: [df[‘日付’].iloc[0]], ‘合計売上’: [df[‘売上’].sum()], ‘平均売上’: [df[‘売上’].mean().round(0)], ‘最大売上’: [df[‘売上’].max()], ‘最小売上’: [df[‘売上’].min()], ‘件数’: [len(df)] }) def save_to_s3_partitioned(df, bucket_name, base_path, date): “””日付でパーティショニングしてS3に保存””” year = date.strftime(‘%Y’) month = date.strftime(‘%m’) day = date.strftime(‘%d’) s3_path = f”{base_path}/year={year}/month={month}/day={day}/summary.csv” csv_buffer = io.StringIO() df.to_csv(csv_buffer, index=False) s3 = boto3.client(‘s3’) s3.put_object( Bucket=bucket_name, Key=s3_path, Body=csv_buffer.getvalue().encode(‘utf-8’) ) return s3_path def run_daily_sales_pipeline(start_date, end_date, bucket_name): “””日次売上パイプラインを実行””” current_date = start_date print(“=” * 60) print(“日次売上パイプライン開始”) print(“=” * 60) while current_date <= end_date: # 1. 売上データ生成 sales_df = generate_daily_sales(current_date) # 2. 集計 summary_df = aggregate_daily_sales(sales_df) # 3. S3に保存 s3_path = save_to_s3_partitioned( summary_df, bucket_name, 'daily-sales-summary', current_date ) total_sales = summary_df['合計売上'].iloc[0] print(f"✅ {current_date.strftime('%Y-%m-%d')}: 合計売上 ¥{total_sales:,}") # 次の日へ current_date += timedelta(days=1) print("=" * 60) print("パイプライン完了！") # 実行例（2024年1月1日〜5日） start = datetime(2024, 1, 1) end = datetime(2024, 1, 5) run_daily_sales_pipeline(start, end, 'my-data-bucket')

# ===== .envファイルの内容（Gitにはコミットしない！）===== AWS_ACCESS_KEY_ID=your_access_key_here AWS_SECRET_ACCESS_KEY=your_secret_key_here AWS_DEFAULT_REGION=ap-northeast-1 S3_BUCKET_NAME=my-data-bucket

# ===== Pythonコード ===== import os from dotenv import load_dotenv import boto3 # .envファイルを読み込み load_dotenv() # 環境変数から取得 BUCKET_NAME = os.getenv(‘S3_BUCKET_NAME’) # S3クライアント作成（自動的に環境変数から認証情報を取得） s3 = boto3.client(‘s3’) print(f”✅ バケット ‘{BUCKET_NAME}’ に接続しました”)

{ “Version”: “2012-10-17”, “Statement”: [ { “Effect”: “Allow”, “Action”: [ “s3:PutObject”, “s3:GetObject”, “s3:ListBucket” ], “Resource”: [ “arn:aws:s3:::my-data-bucket/*”, “arn:aws:s3:::my-data-bucket” ] } ] }