将 Trino 与 Dataproc 搭配使用

---

Trino（以前称为 Presto）是一种分布式 SQL 查询 用于查询分布在一个或多个组织中的大型数据集的引擎 异构数据源。Trino 可以查询 Hive、MySQL、Kafka 和其他数据 来源。本教程将介绍如何执行以下操作：

• 在 Dataproc 集群上安装 Trino 服务
• 从已安装的 Trino 客户端查询公开数据 （位于与 Trino 服务通信的本地机器上） 集群上
• 从与 通过 Trino Java JDBC 驱动程序在集群上启用 Trino 服务。

目标

创建安装了 Trino 的 Dataproc 集群

准备数据。这个 教程使用芝加哥出租车行程 公共数据集，可在 BigQuery 中使用。

1. 从 BigQuery 中提取数据
2. 以 CSV 文件的形式将数据加载到 Cloud Storage 中
3. 转换数据：
   1. 将数据公开为 Hive 外部表， 可由 Trino 查询
   2. 将 CSV 格式的数据转换为 Parquet 格式，以加快查询速度

使用 SSH 隧道或 Trino JDBC 发送 Trino CLI 或应用代码查询 驱动程序分别传递到集群上运行的 Trino 协调器

通过 Trino 网页界面查看日志并监控 Trino 服务 <ph type="x-smartling-placeholder">

</ph>

费用

在本文档中，您将使用 Google Cloud 的以下收费组件：

• Dataproc
• Cloud Storage
• BigQuery

您可使用价格计算器根据您的预计使用情况来估算费用。 Google Cloud 新用户可能有资格申请免费试用。

准备工作

如果您尚未创建 Google Cloud 项目和 用于保存本教程中使用的数据的 Cloud Storage 存储桶。 1. 设置项目

1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. 确保您的 Google Cloud 项目已启用结算功能。

4. 启用 Dataproc, Compute Engine, Cloud Storage, and BigQuery API。

   启用 API

5. 安装 Google Cloud CLI。

6. 如需初始化 gcloud CLI，请运行以下命令：

   gcloud init

7. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

8. 确保您的 Google Cloud 项目已启用结算功能。

9. 启用 Dataproc, Compute Engine, Cloud Storage, and BigQuery API。

   启用 API

10. 安装 Google Cloud CLI。

11. 如需初始化 gcloud CLI，请运行以下命令：

    gcloud init

1.创建 Cloud Storage 存储桶 来保存本教程中使用的数据。

1. 在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。

   进入“存储桶”页面

2. 点击创建存储分区。
3. 在创建存储分区页面上，输入您的存储分区信息。要转到下一步，请点击继续。
   • 在指定存储分区的名称中，输入符合存储分区命名要求的名称。
   • 对于选择数据存储位置，执行以下操作：
     • 选择位置类型选项。
     • 选择位置选项。
   • 对于为数据选择一个默认存储类别，请选择一个存储类别。
   • 对于选择如何控制对象的访问权限，请选择访问权限控制选项。
   • 对于高级设置（可选），请指定加密方法、保留政策或存储分区标签。
4. 点击创建。

创建 Dataproc 集群

通过从本地机器的终端窗口运行此部分中显示的命令来创建集群。

使用 optional-components 标志创建 Dataproc 集群 （适用于映像版本 2.1 及更高版本）以安装 Trino 可选组件 和 enable-component-gateway 标志，以启用 组件网关，以允许 从 Google Cloud 控制台访问 Trino 网页界面。

1. 设置环境变量：
   • PROJECT：您的项目 ID
   • BUCKET_NAME：您在准备工作中创建的 Cloud Storage 存储分区的名称
   • REGION：将在其中创建此教程所使用集群的区域，例如“us-west1”
   • WORKERS：此教程推荐配备 3 到 5 个工作器

   export PROJECT=project-id
   export WORKERS=number
   export REGION=region
   export BUCKET_NAME=bucket-name
   

2. 在本地机器上运行 Google Cloud CLI 以创建集群。

   gcloud beta dataproc clusters create trino-cluster \
       --project=${PROJECT} \
       --region=${REGION} \
       --num-workers=${WORKERS} \
       --scopes=cloud-platform \
       --optional-components=TRINO \
       --image-version=2.1  \
       --enable-component-gateway
   

   创建一个 高可用性集群 - 具有多个主节点的集群 - 不推荐使用， 协调器在主节点 0 上启动， 将保持空闲状态。

准备数据

导出 bigquery-public-data chicago_taxi_trips 以 CSV 文件格式将数据集保存到 Cloud Storage，然后创建一个 Hive 外部表 来引用数据

1. 在本地计算机上，运行以下命令以导入 将 BigQuery 以不含标题的 CSV 文件形式导入 Cloud Storage 您在准备工作中创建的存储桶。

   bq --location=us extract --destination_format=CSV \
        --field_delimiter=',' --print_header=false \
          "bigquery-public-data:chicago_taxi_trips.taxi_trips" \
          gs://${BUCKET_NAME}/chicago_taxi_trips/csv/shard-*.csv
   

2. 创建由您的 Cloud Storage 存储分区中的 CSV 和 Parquet 文件支持的 Hive 外部表。
   1. 创建 Hive 外部表 chicago_taxi_trips_csv。

      gcloud dataproc jobs submit hive \
          --cluster trino-cluster \
          --region=${REGION} \
          --execute "
              CREATE EXTERNAL TABLE chicago_taxi_trips_csv(
                unique_key   STRING,
                taxi_id  STRING,
                trip_start_timestamp  TIMESTAMP,
                trip_end_timestamp  TIMESTAMP,
                trip_seconds  INT,
                trip_miles   FLOAT,
                pickup_census_tract  INT,
                dropoff_census_tract  INT,
                pickup_community_area  INT,
                dropoff_community_area  INT,
                fare  FLOAT,
                tips  FLOAT,
                tolls  FLOAT,
                extras  FLOAT,
                trip_total  FLOAT,
                payment_type  STRING,
                company  STRING,
                pickup_latitude  FLOAT,
                pickup_longitude  FLOAT,
                pickup_location  STRING,
                dropoff_latitude  FLOAT,
                dropoff_longitude  FLOAT,
                dropoff_location  STRING)
              ROW FORMAT DELIMITED
              FIELDS TERMINATED BY ','
              STORED AS TEXTFILE
              location 'gs://${BUCKET_NAME}/chicago_taxi_trips/csv/';"
      

   2. 验证 Hive 外部表的创建过程。
      跳过此步骤以节省时间并减少资源使用量。 如果您的集群具有 3 个或更少的节点，则此查询可能需要几分钟的时间。您可以 跳过此查询步骤，只验证下面最终的 Hive Parquet 的创建过程， 因为该查询步骤要快得多。

      gcloud dataproc jobs submit hive \
          --cluster trino-cluster \
          --region=${REGION} \
          --execute "SELECT COUNT(*) FROM chicago_taxi_trips_csv;"
      

   3. 创建另一个 Hive 外部表“chicago_taxi_trips_parquet” 包含相同的列，但其中数据存储在 Parquet 格式 以提高查询性能。

      gcloud dataproc jobs submit hive \
          --cluster trino-cluster \
          --region=${REGION} \
          --execute "
              CREATE EXTERNAL TABLE chicago_taxi_trips_parquet(
                unique_key   STRING,
                taxi_id  STRING,
                trip_start_timestamp  TIMESTAMP,
                trip_end_timestamp  TIMESTAMP,
                trip_seconds  INT,
                trip_miles   FLOAT,
                pickup_census_tract  INT,
                dropoff_census_tract  INT,
                pickup_community_area  INT,
                dropoff_community_area  INT,
                fare  FLOAT,
                tips  FLOAT,
                tolls  FLOAT,
                extras  FLOAT,
                trip_total  FLOAT,
                payment_type  STRING,
                company  STRING,
                pickup_latitude  FLOAT,
                pickup_longitude  FLOAT,
                pickup_location  STRING,
                dropoff_latitude  FLOAT,
                dropoff_longitude  FLOAT,
                dropoff_location  STRING)
              STORED AS PARQUET
              location 'gs://${BUCKET_NAME}/chicago_taxi_trips/parquet/';"
      

   4. 将 Hive CSV 表中的数据加载到 Hive Parquet 表。

      gcloud dataproc jobs submit hive \
          --cluster trino-cluster \
          --region=${REGION} \
          --execute "
              INSERT OVERWRITE TABLE chicago_taxi_trips_parquet
              SELECT * FROM chicago_taxi_trips_csv;"
      

   5. 验证数据是否已正确加载。

      gcloud dataproc jobs submit hive \
          --cluster trino-cluster \
          --region=${REGION} \
          --execute "SELECT COUNT(*) FROM chicago_taxi_trips_parquet;"
      

运行查询

您可以在本地从 Trino CLI 或应用运行查询。

Trino CLI 查询

本部分演示如何使用 Trino CLI。

1. 在本地机器上运行以下命令，以通过 SSH 连接到集群的主节点。在执行命令期间，本地终端将停止响应。

   gcloud compute ssh trino-cluster-m
   

2. 在集群主节点上的 SSH 终端窗口中，运行 Trino CLI，用于连接到在主实例上运行的 Trino 服务器 节点。

   trino --catalog hive --schema default
   

3. 在 trino:default 提示符处，验证 Trino 是否可以找到 Hive 表。

   show tables;
   

   Table
   ‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐
    chicago_taxi_trips_csv
    chicago_taxi_trips_parquet
   (2 rows)
   

4. 从 trino:default 提示符运行查询，并比较查询 Parquet 与 CSV 数据的性能。
   • Parquet 数据查询

     select count(*) from chicago_taxi_trips_parquet where trip_miles > 50;
     

      _col0
     ‐‐‐‐‐‐‐‐
      117957
     (1 row)
     
     Query 20180928_171735_00006_2sz8c, FINISHED, 3 nodes
     Splits: 308 total, 308 done (100.00%)
     0:16 [113M rows, 297MB] [6.91M rows/s, 18.2MB/s]
     

   • CSV 数据查询

     select count(*) from chicago_taxi_trips_csv where trip_miles > 50;
     

     _col0
     ‐‐‐‐‐‐‐‐
      117957
     (1 row)
     
     Query 20180928_171936_00009_2sz8c, FINISHED, 3 nodes
     Splits: 881 total, 881 done (100.00%)
     0:47 [113M rows, 41.5GB] [2.42M rows/s, 911MB/s]
     

Java 应用查询

如需通过 Trino Java JDBC 驱动程序从 Java 应用运行查询，请执行以下操作： 1.下载 Trino Java JDBC 驱动程序。 1. 在以下代码中添加 trino-jdbc 依赖项 Maven pom.xml。

<dependency>
  <groupId>io.trino</groupId>
  <artifactId>trino-jdbc</artifactId>
  <version>376</version>
</dependency>

Java 代码示例

package dataproc.codelab.trino;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
public class TrinoQuery {
  private static final String URL = "jdbc:trino://trino-cluster-m:8080/hive/default";
  private static final String SOCKS_PROXY = "localhost:1080";
  private static final String USER = "user";
  private static final String QUERY =
      "select count(*) as count from chicago_taxi_trips_parquet where trip_miles > 50";
  public static void main(String[] args) {
    try {
      Properties properties = new Properties();
      properties.setProperty("user", USER);
      properties.setProperty("socksProxy", SOCKS_PROXY);
      Connection connection = DriverManager.getConnection(URL, properties);
      try (Statement stmt = connection.createStatement()) {
        ResultSet rs = stmt.executeQuery(QUERY);
        while (rs.next()) {
          int count = rs.getInt("count");
          System.out.println("The number of long trips: " + count);
        }
      }
    } catch (SQLException e) {
      e.printStackTrace();
    }
  }
}

日志记录和监控

日志记录

Trino 日志位于集群主实例上的 /var/log/trino/ 处， 工作器节点数量。

网页界面

请参阅查看和访问组件网关网址 在本地浏览器中打开集群主节点上运行的 Trino 网页界面。

监控

Trino 通过运行时表公开集群运行时信息。 在 Trino 会话（来自 trino:default）提示中， 运行以下查询以查看运行时表数据：

select * FROM system.runtime.nodes;

清理

完成本教程后，您可以清理您创建的资源，让它们停止使用配额，以免产生费用。以下部分介绍如何删除或关闭这些资源。

删除项目

为了避免产生费用，最简单的方法是删除您为本教程创建的项目。

如需删除项目，请执行以下操作：

警告：删除项目会产生以下影响

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目可以帮助您避免超出项目配额限制。

1. 在 Google Cloud 控制台中，进入管理资源页面。

   转到“管理资源”

2. 在项目列表中，选择要删除的项目，然后点击删除。
3. 在对话框中输入项目 ID，然后点击关闭以删除项目。

删除集群

• 如需删除您的集群，请执行以下操作：

  gcloud dataproc clusters delete --project=${PROJECT} trino-cluster \
      --region=${REGION}
  

删除存储桶

• 要删除您在 Cloud Storage 中创建的 Cloud Storage 存储桶，请执行以下操作： 准备工作，包括数据文件 存储在存储桶中：

  gcloud storage rm gs://${BUCKET_NAME} --recursive