在 Oracle Big Data Service 上将 Apache Impala 脚本迁移到 Trino

简介

Oracle Big Data Service (BDS) 是 Oracle 提供的基于云的服务，用户可以使用它创建和管理 Hadoop 集群、Spark 集群、Trino 和其他大数据服务。在大数据分析领域，Trino 作为一个高性能的分布式 SQL 查询引擎脱颖而出，专为在大型数据集上运行交互式分析查询而设计。Trino 支持在统一的 SQL 接口中跨多个源（包括 Hadoop 分布式文件系统 (HDFS)、OCI 存储桶和传统关系数据库）查询数据。它能够以低延迟处理大规模数据，使其成为数据工程师和分析师的强大工具。

随着向现代数据平台的转变，许多组织正在从 Apache Impala 迁移到 Oracle Big Data Service 上的 Trino。本教程提供了一种逐步的方法来自动和手动调整 Apache Impala 脚本，以便无缝迁移到 Trino。

目标

自动将 Apache Impala 脚本转换为 Trino 兼容 SQL。
突出显示特定方案所需的手动更改。
提供常见迁移挑战的故障排除提示。

先决条件

访问安装了 Trino 的 Oracle Big Data Service。
具备 Apache Impala 和 Trino SQL 的基本知识。
有足够的权限运行脚本和修改配置。

任务 1：自动转换脚本

自动脚本旨在简化将 Apache Impala 脚本转换为与 Oracle Big Data Service 上的 Trino 兼容的过程。此脚本可自动执行多个关键转换，从而减少所需的人工工作量，并尽可能减少迁移过程中出现人为错误的可能性。下面是其功能的细分。

文件处理和初始化。
- 该脚本从读取输入 .hql 文件（通常用于 Apache Impala 查询）开始。
- 它创建一个新的输出文件，其中将保存转换的 Trino 兼容查询。
- 该脚本首先添加一行以使用 USE catalog_name.schema_name 设置目录和数据库上下文，确保查询在 Trino 中的正确上下文中运行。

自动执行查询转换。

覆盖表处理：该脚本在 Apache Impala 脚本中查找任何 OVERWRITE TABLE 语句，并在其上方插入相应的 DELETE FROM 语句。这可确保在插入新数据之前清除表，并与 Trino 处理数据覆盖的方式保持一致。
替换 Impala 特定的函数：该脚本应用多个转换，以将 Apache Impala 特定的语法和函数替换为 Trino 兼容的语法和函数。例如：
- 将注释掉 invalidate metadata 和 REFRESH 命令，因为 Trino 中不需要这些命令。
- 数据类型更改（如 string 到 varchar）是为了匹配 Trino 的数据类型。
- Impala 的 TRUNCATE TABLE 语句将替换为 DELETE FROM，因为 Trino 对截断的处理方式不同。
- COMPUTE STATS 将替换为 ANALYZE，反映 Trino 如何收集统计信息。
- 其他特定的转换包括处理函数，如 nvl、ifnull 和 from_utc_timestamp，将它们调整为 Trino 的语法和函数。
- 将 nvl2 函数转换为等效的 COALESCE 表达式。
- 将正则表达式匹配 (REGEXP) 修改为 Trino 的 REGEXP_LIKE。
- 调整日期和时间戳函数，使其与 Trino 的语法保持一致。

脚本：

#!/bin/bash
# Function to apply transformations to an HQL file
process_file() {
   local input_file="$1"
   local output_file="${input_file%.hql}_modified.hql"
   local diff_file="${input_file%.hql}_diff.sql"

   # Create the output file with the initial modification
   {
      echo "use catalog_name.schema_name;"
      cat "$input_file"
   } > "$output_file"

   # Insert DELETE FROM statements above OVERWRITE TABLE statements
   awk '
   BEGIN { IGNORECASE = 1 }
   {
      if (match($0, /OVERWRITE[[:space:]]+TABLE[[:space:]]+([^[:space:]]+)/, arr)) {
            print "DELETE FROM " arr[1] ";"
      }
      else if (match($0, /OVERWRITE[[:space:]]+([^[:space:]]+)/, arr)) {
      print "DELETE FROM " arr[1] ";"
      }
      print
   }' "$output_file" > temp_file && mv temp_file "$output_file"

   # Define replacements as array of tuples (search, replace)
   declare -a replacements=(
      "old_line1 new_line1"
      "old_line2 new_line2"
      # Add more replacements as needed
   )

   # Apply replacements
   for replacement in "${replacements[@]}"; do
      search_text=$(echo "$replacement" | awk '{print $1}')
      replace_text=$(echo "$replacement" | awk '{print $2}')
      sed -i "s/${search_text}/${replace_text}/Ig" "$output_file"
   done

   # Handle specific replacements
   sed -i 's/invalidate metadata/--invalidate metadata/Ig' "$output_file"
   sed -i 's/REFRESH/--REFRESH/Ig' "$output_file"
   sed -i 's/ string/ varchar/Ig' "$output_file"
   sed -i 's/TRUNCATE[[:space:]]\+TABLE[[:space:]]*/DELETE FROM /Ig' "$output_file"
   sed -i 's/TRUNCATE /DELETE FROM /Ig' "$output_file"
   sed -i 's/OVERWRITE[[:space:]]\+TABLE[[:space:]]*/INTO /Ig' "$output_file"
   sed -i 's/COMPUTE[[:space:]]\+stats/ANALYZE/Ig' "$output_file"
   sed -i 's/INSERT[[:space:]]\+INTO[[:space:]]\+TABLE/INSERT INTO/Ig' "$output_file"
   sed -i 's/from_utc_timestamp(now(),/current_timestamp AT TIME ZONE /Ig' "$output_file"
   sed -i "s/'America\/New_York')/'America\/New_York'/Ig" "$output_file"
   sed -i 's|/\* +NOSHUFFLE \*/| |Ig' "$output_file"
   sed -i 's/into[[:space:]]\+table/into/Ig' "$output_file"
   sed -i 's/STRAIGHT_JOIN/ /Ig' "$output_file"
   sed -i 's/OVERWRITE[[:space:]]\+/INTO /Ig' "$output_file"
   sed -i 's/TABLE[[:space:]]\+INTO/INTO/Ig' "$output_file"
   sed -i 's/SET[[:space:]]\+DECIMAL_V2/--SET DECIMAL_V2/Ig' "$output_file"
   sed -i 's/nvl(/COALESCE(/Ig' "$output_file"
   sed -i 's/ifnull(/COALESCE(/Ig' "$output_file"
   sed -i 's/CREATE[[:space:]]\+EXTERNAL/CREATE/Ig' "$output_file"
   sed -i -E 's/TRUNC[[:space:]]*\(/TRUNCATE(/Ig' "$output_file"
   sed -i -E 's/CAST[[:space:]]*\(/TRY_CAST(/Ig' "$output_file"
   sed -i -E 's/instr[[:space:]]*\(/strpos(/Ig' "$output_file"
   sed -i 's/""[[:space:]]\+as/null as/Ig' "$output_file"
   sed -i -E 's/\$\{var\:([^}]*)\}/\$\{\1\}/Ig' "$output_file"
   sed -i -E 's/([a-zA-Z_][a-zA-Z0-9_]*)[[:space:]]+REGEXP[[:space:]]+('\''[^'\'']*'\'')/REGEXP_LIKE(\1, \2)/Ig' "$output_file"
   sed -i -E 's/lower\(([a-zA-Z_][a-zA-Z0-9_]*\))[[:space:]]+REGEXP[[:space:]]+('\''[^'\'']*'\'')/REGEXP_LIKE(lower(\1, \2)/Ig' "$output_file"
   sed -i "s/)\s*STORED\s*AS\s*PARQUET\s*LOCATION\s*'\([^']*\)';/) WITH (\nexternal_location = '\1',\nformat = 'PARQUET'\n);/Ig" "$output_file"    
   sed -i -E "s/try_cast\(extract\(([a-zA-Z0-9_.]+), 'MONTH'\) AS varchar\)/try_cast(extract(MONTH from try_cast(\1 AS timestamp)) as VARCHAR)/Ig" "$output_file"
   sed -i -E "s/TRY_CAST\(extract\(add_months\(([^,]+),([^,]+)\),[[:space:]]*'MONTH'\)[[:space:]]*AS[[:space:]]*varchar\)/TRY_CAST\(extract\(MONTH from date_add\('month',\2,try_cast\(\1 as timestamp\)\)\) AS varchar\)/Ig" "$output_file"
   sed -Ei "s/try_cast\(ADD_MONTHS\(([^,]+),\s*-?\s*([0-9]+)\) AS varchar\)/try_cast\(date_add\('month', -\2, try_cast(\1 as timestamp)\) AS varchar\)/Ig" "$output_file"
   sed -i -E 's/DATEDIFF\(/date_diff\('\''day'\'', /Ig' "$output_file"
   sed -i -E 's/TO_DATE\(/DATE\(/Ig' "$output_file"
   sed -i -E 's/now\(\)/current_timestamp/Ig' "$output_file"

   # Generate the diff file
   diff -u "$input_file" "$output_file" > "$diff_file"

   echo "Processed $input_file -> $output_file. Diff file created: $diff_file"
}

# Check arguments
if [ "$#" -ne 1 ]; then
   echo "Usage: $0 <file_or_directory>"
   exit 1
fi

TARGET="$1"

# Check if the target is a directory
if [ -d "$TARGET" ]; then
   for file in "$TARGET"/*.hql; do
      [ -e "$file" ] || continue # Skip if no HQL files found
      process_file "$file"
   done
elif [ -f "$TARGET" ]; then
   process_file "$TARGET"
else
   echo "Invalid input: $TARGET is neither a file nor a directory."
   exit 1
fi

注意：此脚本提供了将 Apache Impala 脚本转换为 Trino 的基础方法，但未涵盖所有可能的 Impala 语句和方案。如果您遇到此脚本未处理的其他特定于 Impala 的语法或函数，则可以通过添加新的转换规则来扩展该脚本以涵盖这些情况。通过这些修改定期更新脚本将提高其未来转换的有效性

任务 2：运行转换脚本

将脚本保存到 Oracle Big Data Service 边缘节点中。
使用包含 Apache Impala 脚本的目标文件或目录运行脚本。
查看生成的输出和差异文件以进行更改。

任务 3：进行人工调整

使用 LEFT ANTI JOIN 语句。

某些 Apache Impala SQL 结构（如 LEFT ANTI JOIN）需要手动干预。确定 Impala 脚本中的 LEFT ANTI JOIN 语句，并使用 LEFT JOIN 的等效 Trino SQL 将它们替换为 WHERE 子句。

脚本示例：
- Impala 脚本：
```
SELECT t1.*
 FROM table1 t1
 LEFT ANTI JOIN table2 t2 ON t1.key = t2.key;
```
- Trino 脚本：
```
SELECT t1.*
 FROM table1 t1
 LEFT JOIN table2 t2 ON t1.key = t2.key
 WHERE t2.key IS NULL;
```

调整 GROUP BY 子句。

Trino 允许在 GROUP BY 子句中基于位置的分组，这可以简化长分组。如有必要，将列名转换为 GROUP BY 子句中的等效位置。确保所有列都已正确映射。

脚本示例：

Impala 脚本：

GROUP BY sales_data.region,
    sales_data.product_name,
    sales_data.transaction_id,
    transaction_date,
    last_updated,
    inventory.stock_quantity,
    sales_amount,
    discount_status,
    inventory_cost,
    total_revenue,
    revenue_status,
    reporting_period,
    report_generated_date;

Trino 脚本：

GROUP BY sales_data.region,
    sales_data.product_name,
    sales_data.transaction_id,
    transaction_date,
    last_updated,
    inventory.stock_quantity,
    7,  -- sales_amount
    8,  -- discount_status
    9,  -- inventory_cost
    10, -- total_revenue
    11, -- revenue_status
    12, -- reporting_period
    13; -- report_generated_date

故障诊断和技巧

错误处理：如果在转换后遇到语法错误，请仔细查看脚本，特别是有关 JOIN 子句和日期函数的脚本。
测试：在 Trino 中验证查询的输出，以确保它们与 Apache Impala 的预期结果匹配。
请根据需要随时参考 Trino 文档。

后续步骤

在完成将 Apache Impala 脚本迁移到 Oracle Big Data Service 上的 Trino 之后，必须彻底验证转换的脚本。首先在测试环境中运行转换后的 Trino 脚本，以确保它们产生预期的结果。密切关注性能差异，特别是在涉及联接或聚合的复杂查询中，因为 Trino 的执行引擎与 Impala 的行为可能不同。

此外，考虑通过分析查询计划并进行必要的调整来优化新的 Trino 查询。最后，记录在迁移过程中学到的任何具体挑战或经验教训，因为这对未来的迁移很有价值。

致谢

Authors — Pavan Upadhyay（首席云工程师）、Saket Bihari（首席云工程师）