@dsl.pipeline(
    name="Ingest data",
    description="Ingest data from S3."
)
def ingest_pipeline(csv_path: str):
  # TODO: define ContainerOp to accsss S3 bucket

kfp.compiler.Compiler().compile(
    pipeline_func=ingest_pipeline,
    package_path="ingest_pipeline.yaml"
)

# S3 のシークレット情報
# ACCESS_KEY, SECRET_ACCESS_KEY は base64に変換したものを指定する
# ex) `echo -n $AWS_ACCESS_KEY_ID | base64`
apiVersion: v1
kind: Secret
metadata:
  name: aws-secret
  namespace: ***
type: Opaque
data:
  AWS_ACCESS_KEY_ID: *****
  AWS_SECRET_ACCESS_KEY: *****

def prepare(bucket_name: str, input_file_name: str):
  from io import StringIO

  import boto3
  import pandas as pd

  client = boto3.client("s3")
  obj = client.get_object(Bucket=bucket_name, Key=input_file_name)
  content = obj["Body"].read().decode("utf-8")

  df = pd.read_csv(StringIO(content))
  df.head()

from kfp.aws import use_aws_secret

prepare_op = comp.create_component_from_func(
    prepare,
    packages_to_install=["boto3", "pandas"]
)

@dsl.pipeline(
    name="Ingest data",
    description="Ingest data from S3."
)
def ingest_pipeline(bucket_name: str, csv_path: str):
  prepare_op(bucket_name=bucket_name, input_file_name=csv_path).apply(
    use_aws_secret(
        'aws-secret', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY'
    )
  )

kfp.compiler.Compiler().compile(
    pipeline_func=ingest_pipeline,
    package_path="ingest_pipeline.yaml"
)