如何通过Python在Trino中执行查询并生成加密Avro格式输出？

一、先排查Avro格式导出报错的常见原因 #

1. 依赖版本不兼容问题 #

Python连接Trino常用trino库，导出Avro需确保依赖包版本匹配，版本不匹配是高频报错原因：

• 安装指定版本依赖：

  pip install avro==1.11.0 trino==0.320.0
  

• 同时确认Trino服务器端的Avro插件版本和客户端依赖对应，低版本插件可能不支持Decimal、复杂数组等数据类型。

2. 数据类型适配处理 #

Trino返回的Decimal、Array、Map等类型直接转Avro容易报错，需手动转换：

from trino.dbapi import connect
import avro.schema
from avro.datafile import DataFileWriter
from avro.io import DatumWriter
import decimal

# 连接Trino
conn = connect(
    host='你的Trino地址',
    port=8080,
    user='你的用户名',
    catalog='hive',
    schema='目标库'
)

cursor = conn.cursor()
cursor.execute("SELECT * FROM 你的表名 LIMIT 100")
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]

# 处理Decimal类型（转字符串适配Avro）
processed_rows = []
for row in rows:
    new_row = []
    for val in row:
        if isinstance(val, decimal.Decimal):
            new_row.append(str(val))
        else:
            new_row.append(val)
    processed_rows.append(dict(zip(columns, new_row)))

# 定义Avro Schema（必须包含null类型兼容空值）
schema = avro.schema.parse("""
{
  "type": "record",
  "name": "HiveTableRecord",
  "fields": [
    {"name": "col1", "type": ["null", "string"]},
    {"name": "col2", "type": ["null", "int"]},
    {"name": "col_decimal", "type": ["null", "string"]}
  ]
}
""")

# 写入Avro文件
with open('output.avro', 'wb') as f:
    writer = DataFileWriter(f, DatumWriter(), schema)
    for row in processed_rows:
        writer.append(row)
    writer.close()

3. 空值兼容设置 #

Avro Schema中每个字段必须显式允许null类型，比如写成["null", "string"]，而非直接"string"，否则遇到空值直接报错。

二、生成加密的Avro文件 #

1. 整体文件加密（AES） #

用pycryptodome对生成的Avro文件做整体加密：

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
import os

# 生成AES密钥和初始化向量（实际场景密钥需安全存储）
key = os.urandom(32)  # AES-256
iv = os.urandom(16)

# 读取Avro原始文件
with open('output.avro', 'rb') as f:
    avro_data = f.read()

# AES-CBC模式加密
cipher = AES.new(key, AES.MODE_CBC, iv)
encrypted_data = cipher.encrypt(pad(avro_data, AES.block_size))

# 保存加密文件（把iv写入文件开头，解密时需要读取）
with open('encrypted_output.avro', 'wb') as f:
    f.write(iv + encrypted_data)

# 保存密钥（实际用密钥管理服务存储，不要明文存本地）
with open('aes_key.bin', 'wb') as f:
    f.write(key)

2. 字段级加密（针对敏感字段） #

如果只需加密部分敏感字段，可在写入Avro前先加密字段值：

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
import base64

key = os.urandom(32)
iv = os.urandom(16)
cipher = AES.new(key, AES.MODE_CBC, iv)

# 字段加密函数
def encrypt_field(value):
    if value is None:
        return None
    encrypted = cipher.encrypt(pad(str(value).encode('utf-8'), AES.block_size))
    return base64.b64encode(encrypted).decode('utf-8')

# 处理敏感字段加密
processed_rows = []
for row in rows:
    new_row = dict(zip(columns, row))
    new_row['敏感字段名'] = encrypt_field(new_row['敏感字段名'])
    processed_rows.append(new_row)

# 后续写入Avro文件逻辑同前

三、额外注意事项 #

• 若用Trino原生SELECT ... INTO OUTFILE导出Avro，需确保服务器端配置Avro输出支持，且目标路径有权限，加密可通过服务器端加密文件系统实现。
• 复杂类型（如Struct、Array）的Avro Schema需精准定义，比如Struct对应Avro的record类型。

内容的提问来源于stack exchange，提问作者shashankS