Apache Seatunnel & HAQM Bedrock 助力生成式 AI 应用规模化生产实践

{"Item":{"review_id":{"S":"AEEZL8Z5691IJ"},"date":{"S":"1215475200"},"customer":{"S":"HAQM Customer"},"asin":{"S":"B000Q6R4MK"},"review":{"S":"I can hear the caller just great -- but I get frequent \"what?\" \"I can't hear you\"  \"it sounds like a lot of background noise\" , etc. comments.  I'm trying to figure out the best way to have it arranged on my visor, etc.  Not 100% sold yet."},"rating":{"S":"4.0"}}}

env {
  # Set the execution engine to SeaTunnel Zeta Engine
  execution.engine = "seatunnel"

  # Set job mode to BATCH for processing the JSON file
  job.mode = "BATCH"
}

source {
  S3File {
    path = "/data/3vk7gdzq6myxhn2kwexoiywjh4.json"
    bucket = "s3a://opensearch"
    file_format_type = "json"

    # AWS region configuration
    fs.s3a.endpoint = "s3.us-east-1.amazonaws.com"

    # Use SimpleAWSCredentialsProvider instead of InstanceProfileCredentialsProvider
    fs.s3a.aws.credentials.provider = "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"

    # Provide explicit AWS credentials - 请替换为您的实际访问密钥
    access_key = ""
    secret_key = ""

    # Additional S3A configuration options
    hadoop_s3_properties {
      "fs.s3a.impl" = "org.apache.hadoop.fs.s3a.S3AFileSystem"
      "fs.s3a.connection.ssl.enabled" = "true"
    }

    # Required schema for JSON file format
    schema  = {
      fields {
        Item = {
            review_id = {
                S = string
            }
            date = {
                S = string
            }
            customer = {
                S = string
            }
            asin = {
                S = string
            }
            review = {
                S = string
            }
            rating = {
                S = string
            }
        }
      }
    }

    # Register output table
    plugin_output = "s3_data"
  }
}

transform {
  # First transform to extract the actual review text from the nested S field
  Sql {
    plugin_input = "s3_data"
    plugin_output = "extracted_data"

    query = "SELECT Item.review_id.S as review_id, Item.date.S as date, Item.customer.S as customer, Item.asin.S as asin, Item.review.S as review, Item.rating.S as rating FROM s3_data"
  }

  # Use HAQM Bedrock to generate embeddings for the review field
  Embedding {
    plugin_input = "extracted_data"
    plugin_output = "embedded_data"

    # Specify the model provider as AMAZON for Bedrock
    model_provider = "AMAZON"

    # Specify the model ID for HAQM Titan Embeddings
    model = "amazon.titan-embed-text-v2:0"

    # AWS region for Bedrock service
    region = "us-east-1"

    # AWS credentials for Bedrock service - 请替换为您的实际访问密钥
    api_key = ""
    secret_key = ""

    # Define which fields to vectorize and their target fields
    vectorization_fields {
      review_embedding = review
    }

    # Batch size for processing
    single_vectorized_input_number = 10
    dimension = 1024
  }
}

sink {
  Console {
    plugin_input = "embedded_data"  # 使用 Embedding 转换后的数据

    # 只打印少量记录以便于查看
    limit = 10
  }
  Elasticsearch {
    plugin_input = "embedded_data"

    # OpenSearch endpoint
    hosts = ["http://xxxxxx.us-east-1.es.amazonaws.com"]
    tls_verify_certificate = false
    # Index configuration
    index = "reviews"
    username = ""
    password = ""
    vectorization_fields = ["review_embedding"]
    # 指定向量维度（每个向量中的浮点数数量）
    vector_dimensions = 1024

  }
}

GET reviews/_search
{
  "size": 5,
  "query": {
    "neural": {
      "review_embedding": {
        "query_text": "Installed and connected pretty well. Works good and keeps my eyes on the road. Took a few weeks to see the on and off button, but overall I like this and would suggest it.",
        "model_id": "8xwrFJYB5648rVcWvwIU",
        "k": 10
      }
    }
  }
}