将 Trino 与 Dataproc 搭配使用

---

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

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

目标

创建 Dataproc 集群并安装 Trino

准备数据。本教程使用 BigQuery 中提供的 Chicago Taxi Trips（芝加哥出租车行程）公共数据集。

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

向集群上运行的 Trino 协调器发送 Trino CLI 查询（通过 SSH 隧道发送）或应用代码查询（通过 Trino JDBC 驱动程序发送）

通过 Trino 网页界面检查日志并监控 Trino 服务

费用

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

• Dataproc
• Cloud Storage
• BigQuery

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

准备工作

如果您尚未执行此操作，请创建 Google Cloud 项目和 Cloud Storage 存储桶以保留此教程中使用的数据。1. 设置项目

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

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. Make sure that billing is enabled for your Google Cloud project.

4. Enable the Dataproc, Compute Engine, Cloud Storage, and BigQuery APIs.

   Enable the APIs

5. Install the Google Cloud CLI.

6. To initialize the gcloud CLI, run the following command:

   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. Make sure that billing is enabled for your Google Cloud project.

9. Enable the Dataproc, Compute Engine, Cloud Storage, and BigQuery APIs.

   Enable the APIs

10. Install the Google Cloud CLI.

11. To initialize the gcloud CLI, run the following command:

    gcloud init

1. 在项目中创建 Cloud Storage 存储桶以保留此教程中使用的数据。

1. In the Google Cloud console, go to the Cloud Storage Buckets page.

   Go to Buckets page

2. Click Create bucket.
3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
   • For Name your bucket, enter a name that meets the bucket naming requirements.
   • For Choose where to store your data, do the following:
     • Select a Location type option.
     • Select a Location option.
   • For Choose a default storage class for your data, select a storage class.
   • For Choose how to control access to objects, select an Access control option.
   • For Advanced settings (optional), specify an encryption method, a retention policy, or bucket labels.
4. Click Create.

创建 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. 在本地机器上，运行以下命令，以 CSV 文件（不含标题）形式将出租车数据从 BigQuery 导入您在准备工作中创建的 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 查询 Hive Parquet 出租车数据集。

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. 在 Maven pom.xml 中添加 trino-jdbc 依赖项。

<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. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

删除集群

• 如需删除您的集群，请输入以下命令：

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

删除存储桶

• 如需删除您在准备工作中创建的 Cloud Storage 存储分区（包括存储在存储分区中的数据文件），请输入以下命令：

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