Samrose-Ahmed

Query:
  Summary:
     - Query ID: 5e2bb5da-3fc3-11ef-abdf-0edc7e62e9bc
     - Start Time: 2024-07-11 13:23:00
     - End Time: 2024-07-11 13:23:05
     - Total: 5s588ms
     - Query Type: Query
     - Query State: Finished
     - StarRocks Version: 3.3.0-19a3f66
     - User: root

Samrose-Ahmed

operator:
  
  global:
    rbac:
      create: true
      serviceAccount:
        name: "starrocks"
        # Optional annotations to add to serviceaccount manifest
        annotations: {}

Samrose-Ahmed

/*-
 * #%L
 * athena-example
 * %%
 * Copyright (C) 2019 Amazon Web Services
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *

Samrose-Ahmed

/*-
 * #%L
 * athena-example
 * %%
 * Copyright (C) 2019 Amazon Web Services
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *

Samrose-Ahmed

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.from_json

Samrose-Ahmed

fun checkDuplicate(sequencer: String): Boolean {
    // TTL to expire old DynamoDB items
    val expireTime = ((System.currentTimeMillis() / 1000L) + DDB_ITEM_EXPIRE_SECONDS).toString()
    val attrs = mapOf(
            "sequencer" to AttributeValue(sequencer),
            "ttl" to AttributeValue().apply { this.setN(expireTime) }
    )
    val req = PutItemRequest(DUPLICATES_DDB_TABLE_NAME, attrs)
            .apply { this.conditionExpression = "attribute_not_exists(sequencer)" }

Samrose-Ahmed

fun readParquetMetrics(s3Path: String, table: Table): Metrics {
    return ParquetUtil.fileMetrics(fileIO.newInputFile(s3Path), MetricsConfig.forTable(table))
}

Samrose-Ahmed

fun processRecord(sqsMessage: SQSMessage): Unit {
    val record = S3EventNotification.parseJson(sqsMessage.body).records[0]
    val s3Bucket = record.s3.bucket.name
    val s3Object = record.s3.`object`
    val s3ObjectKey = s3Object.key
    val s3ObjectSize = s3Object.sizeAsLong
    val s3Path = "s3://$s3Bucket/$s3ObjectKey"

    if (checkDuplicate(s3Object.sequencer)) {
        println("Found duplicate SQS message for key: ${s3ObjectKey}. Skipping...")

Samrose-Ahmed

class IcebergMetadataWriter {
    // In a real world usecase, you can create this dynamically from the data.
    val icebergTable: Table = icebergCatalog.loadTable(TableIdentifier.of(Namespace.of(NAMESPACE), TABLE_NAME))
    val appendFiles: AppendFiles = icebergTable.newAppend()

    // Lambda handler
    fun handle(sqsEvent: SQSEvent) {
        for (record in sqsEvent.records) {
            processRecord(record, tableObjs)
        }

Samrose-Ahmed

{
  "expiration": "2022-02-14T13:08:46.864Z",
  "conditions": [
    { "acl": "bucket-owner-full-control" },
    { "bucket": "my-bucket" },
    ["starts-with", "$key", "stuff/clientId"],
    ["content-length-range", 1048576, 10485760]
  ]
}