在 Dataproc 叢集上執行 Hadoop 字數工作

1. 簡介

工作流程是資料分析的常見用途，內含擷取、轉換與分析資料，藉此找出有參考價值的資訊。在 Google Cloud Platform 中，Cloud Composer 是用於自動調度工作流程的工具，也是熱門開放原始碼工作流程工具 Apache Airflow 的託管版本。在本研究室中，您將使用 Cloud Composer 建立簡單的工作流程，藉此建立 Cloud Dataproc 叢集，並使用 Cloud Dataproc 和 Apache Hadoop 加以分析，並在之後刪除 Cloud Dataproc 叢集。

什麼是 Cloud Composer？

Cloud Composer 是一項全代管工作流程自動化調度管理服務，能讓您建立、排程和監控散布於不同雲端和內部部署資料中心的管道。Cloud Composer 以熱門的 Apache Airflow 開放原始碼專案為基礎，並採用 Python 程式語言進行操作，不受限於單一架構，而且相當容易使用。

使用 Cloud Composer 而非 Apache Airflow 本機執行個體，使用者無須安裝或管理負擔，就能充分運用 Airflow 的優勢。

如要進一步瞭解 Cloud Composer，請參閱說明文件！

什麼是 Apache Airflow？

Apache Airflow 是開放原始碼工具，可用於以程式輔助方式編寫、排程及監控工作流程。您會在本研究室中看到一些與 Airflow 相關的重要詞彙：

• DAG：DAG (有向非循環圖) 是您要排程及執行的經過整理的工作集合。DAG (也稱為工作流程) 是在標準 Python 檔案中定義
• 運算子：運算子會說明工作流程中的單一工作

什麼是 Cloud Dataproc？

Cloud Dataproc 是 Google Cloud Platform 的全代管 Apache Spark 和 Apache Hadoop 服務。Cloud Dataproc 可輕鬆整合其他 GCP 服務，是一款功能強大且完善的平台，可用於資料處理、數據分析和機器學習。

如要進一步瞭解 Cloud Dataproc，請參閱說明文件！

操作步驟

本程式碼研究室說明如何在 Cloud Composer 中建立及執行 Apache Airflow 工作流程，以完成下列工作：

• 建立 Cloud Dataproc 叢集
• 在叢集中執行 Apache Hadoop 字數計算工作，並將結果輸出至 Cloud Storage
• 刪除叢集

課程內容

• 如何在 Cloud Composer 中建立及執行 Apache Airflow 工作流程
• 如何使用 Cloud Composer 和 Cloud Dataproc 對資料集執行分析
• 如何透過 Google Cloud Platform 主控台、Cloud SDK 和 Airflow 網頁介面存取 Cloud Composer 環境

軟硬體需求

• GCP 帳戶
• 具備基本 CLI 知識
• 對 Python 有基本瞭解

2. 設定 GCP

建立專案

選取或建立 Google Cloud Platform 專案。

請記下專案 ID，後續步驟將會用到。

建立新專案時，專案 ID 會顯示在建立頁面的「專案名稱」下方
如果您已建立專案，即可在「專案資訊卡」的控制台首頁中找到專案 ID

注意：如果您不打算保留在本教學課程中建立的資源，請建立新專案，不要選取現有專案。這樣在課程結束後，您就可以刪除專案，移除與專案和教學課程相關聯的所有資源。

啟用 API

啟用 Cloud Composer、Cloud Dataproc 和 Cloud Storage API。啟用這些 API 後，您可以忽略「前往憑證」按鈕，並繼續進行教學課程的下一個步驟。

如要進一步瞭解如何啟用及停用 API，請參閱說明文件

建立 Composer 環境

使用下列設定建立 Cloud Composer 環境： 名稱：my-composer-environment 位置：us-central1 可用區：us-central1-a 所有其他設定均可保留預設設定。按一下底部的「建立」。

如果在 GCP 主控台的「環境」頁面中，環境名稱左側顯示綠色勾號，表示環境已建立完成。建立環境需要幾分鐘的時間，您可以在這段期間離開這個頁面。

建立 Cloud Storage 值區

在專案中，使用下列設定建立 Cloud Storage 值區： 名稱：<your-project-id> 預設儲存空間級別：多區域 地點：美國 存取權控管模型：精細 準備就緒後，按下「建立」

3. 設定 Apache Airflow

如果您尚未完成 Composer 環境的建立作業，請先略過「範例工作流程」步驟，完成後再返回本部分

檢視 Composer 環境資訊

在 GCP Console 中，開啟「Environments」(環境) 頁面

按一下環境名稱即可查看詳細資料。

「環境詳細資料」頁面提供各種資訊，例如 Airflow 網路介面網址、Google Kubernetes Engine 叢集 ID、Cloud Storage 值區名稱，以及 /dags 資料夾的路徑。

在 Airflow 中，DAG (有向非循環圖) 是您要排程及執行的經過整理的工作集合。DAG 也稱為工作流程，是在標準 Python 檔案中定義。Cloud Composer 只會排定 /dags 資料夾中的 DAG。/dags 資料夾位於 Cloud Composer 在您建立環境時自動建立的 Cloud Storage 值區中。

設定 Apache Airflow 環境變數

Apache Airflow 變數是 Airflow 的專屬概念，與環境變數不同。在這個步驟中，您將設定下列三個 Airflow 變數：gcp_project、gcs_bucket 和 gce_zone。

使用 gcloud 設定變數

首先，請開啟 Cloud Shell，已輕鬆安裝 Cloud SDK。

將環境變數 COMPOSER_INSTANCE 設為 Composer 環境名稱

COMPOSER_INSTANCE=my-composer-environment

如要透過 gcloud 指令列工具設定 Airflow 變數，請搭配 variables 子指令使用 gcloud composer environments run 指令。這個 gcloud composer 指令會執行 Airflow CLI 子指令 variables。子指令會將引數傳遞至 gcloud 指令列工具。

您將執行這個指令三次，將變數替換為與專案相關的變數。

使用下列指令設定 gcp_project，將 <your-project-id> 替換為您在步驟 2 中記下的專案 ID。

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcp_project <your-project-id>

輸出內容應如下所示

kubeconfig entry generated for us-central1-my-composer-env-123abc-gke.
Executing within the following Kubernetes cluster namespace: composer-1-10-0-airflow-1-10-2-123abc
[2020-04-17 20:42:49,713] {settings.py:176} INFO - settings.configure_orm(): Using pool settings. pool_size=5, pool_recycle=1800, pid=449
[2020-04-17 20:42:50,123] {default_celery.py:90} WARNING - You have configured a result_backend of redis://airflow-redis-service.default.svc.cluste
r.local:6379/0, it is highly recommended to use an alternative result_backend (i.e. a database).
[2020-04-17 20:42:50,127] {__init__.py:51} INFO - Using executor CeleryExecutor
[2020-04-17 20:42:50,433] {app.py:52} WARNING - Using default Composer Environment Variables. Overrides have not been applied.
[2020-04-17 20:42:50,440] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg
[2020-04-17 20:42:50,452] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg

使用下列指令設定 gcs_bucket，並將 <your-bucket-name> 替換為您在步驟 2 中記下的值區 ID。如果遵循我們的建議，您的值區名稱會與專案 ID 相同。輸出內容會與上一個指令類似，

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcs_bucket gs://<your-bucket-name>

使用下列指令設定 gce_zone。輸出內容會與先前的指令類似。

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gce_zone us-central1-a

(選用) 使用 gcloud 查看變數

如要查看變數的值，請使用 get 引數執行 Airflow CLI 子指令 variables，或使用 Airflow UI。

例如：

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --get gcs_bucket

您可以使用您剛剛設定的三個變數來測試：gcp_project、gcs_bucket 和 gce_zone。

4. 工作流程範例

接著來看看步驟 5 中要使用的 DAG 程式碼。別擔心，請先按照這裡的步驟操作。

這裡有很多要說明的內容，因此讓我們逐一說明。

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

我們先從 Airflow 匯入作業開始：

• airflow.models - 允許我們在 Airflow 資料庫中存取及建立資料。
• airflow.contrib.operators - 社區業者的居住地。在這種情況下，我們需要 dataproc_operator 才能存取 Cloud Dataproc API。
• airflow.utils.trigger_rule：用於在運算子中加入觸發條件規則。觸發條件規則可讓您精細地控管運算子是否應根據其父項狀態執行。

output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep

這會指定輸出檔案的位置。這裡值得注意的是 models.Variable.get('gcs_bucket') 這行，它會從 Airflow 資料庫中擷取 gcs_bucket 變數值。

WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)

input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

• WORDCOUNT_JAR - 我們最終會在 Cloud Dataproc 叢集中執行的 .jar 檔案位置。並會由 GCP 代管。
• input_file - 儲存 Hadoop 工作資料的檔案所在位置。我們會在步驟 5 中一起將資料上傳至該位置。
• wordcount_args：我們會傳入 jar 檔案的引數。

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

如此一來，我們就能得到相當於前一天午夜的日期時間物件。例如，如果此要求是在 3 月 4 日 11:00 執行，則日期時間物件表示 3 月 3 日的 00:00。這與 Airflow 處理排程的方式有關。詳情請參閱這篇文章。

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

每次建立新的 DAG 時，都應提供字典形式的 default_dag_args 變數：

• 'email_on_failure'：指出是否應在工作失敗時傳送電子郵件快訊
• 'email_on_retry'：指出是否應在重試工作時傳送電子郵件快訊
• 'retries' - 表示在 DAG 故障的情況下， Airflow 應執行的重試次數
• 'retry_delay'：表示 Airflow 應等待多久才嘗試重試
• 'project_id' - 告知 DAG 要與哪個 GCP 專案 ID 建立關聯，稍後與 Dataproc 操作者需要使用該 ID

with models.DAG(
        'composer_hadoop_tutorial',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

使用 with models.DAG 會指示指令碼在同一個 DAG 中納入其下的所有內容。我們也看到傳入了三個引數：

• 第一個字串是我們要為 DAG 指定的名稱。在本例中，我們將使用 composer_hadoop_tutorial。
• schedule_interval：datetime.timedelta 物件，我們已將其設為一天。也就是說，這個 DAG 會在 'default_dag_args' 中先前設定的 'start_date' 後，嘗試每天執行一次
• default_args：我們先前建立的字典，其中包含 DAG 的預設引數

建立 Dataproc 叢集

接下來，請建立 dataproc_operator.DataprocClusterCreateOperator 來建立 Cloud Dataproc 叢集。

    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

在這個運算子中，雖然我們看到幾個引數，但第一個引數專屬於這個運算子：

• task_id：就像在 BashOperator 中一樣，這是我們指派給運算子的名稱，可透過 Airflow UI 查看
• cluster_name：我們指派給 Cloud Dataproc 叢集的名稱。我們將其命名為 composer-hadoop-tutorial-cluster-{{ ds_nodash }} (如需其他資訊，請參閱資訊方塊)
• num_workers - 我們為 Cloud Dataproc 叢集分配的工作站數量
• zone：希望叢集儲存的地理區域，儲存在 Airflow 資料庫中。這會讀取我們在步驟 3 中設定的 'gce_zone' 變數
• master_machine_type - 我們要為 Cloud Dataproc 主機分配的機器類型
• worker_machine_type：要分配給 Cloud Dataproc 工作站的機器類型

cluster_name 變數的其他資訊

之所以有參數的 {{ ds_nodash }} 部分，是因為 Airflow 支援 jinja2 範本。運算子是每次啟動運算子時， Airflow 會在執行階段轉譯的參數。在這種情況下，系統會將 {{ ds_nodash }} 替換為 DAG 的執行日期 (格式為 YYYYMMDD)。如果你不熟悉 jinja2 範本，沒關係！您不需要具備相關知識，也能完成本實驗室。

提交 Apache Hadoop 工作

dataproc_operator.DataProcHadoopOperator 可讓我們將工作提交至 Cloud Dataproc 叢集。

    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

我們提供多個參數：

• task_id：我們為 DAG 這部分指派的名稱
• main_jar - 我們要對叢集執行的 .jar 檔案的位置
• cluster_name - 要用來執行工作的叢集名稱，您會注意到與前一個運算子中找到的相同
• arguments - 傳遞至 jar 檔案的引數，就像從指令列執行 .jar 檔案一樣

刪除叢集

最後一個要建立的運算子是 dataproc_operator.DataprocClusterDeleteOperator。

    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

顧名思義，這個運算子會刪除指定的 Cloud Dataproc 叢集。我們在這裡看到三個引數：

• task_id - 就像在 BashOperator 中一樣，這是您指派給運算子的名稱，該名稱可透過 Airflow UI 查看
• cluster_name：我們為 Cloud Dataproc 叢集指派的名稱。我們在此將其命名為 composer-hadoop-tutorial-cluster-{{ ds_nodash }} (如需其他選用資訊，請參閱「建立 Dataproc 叢集」後方的資訊方塊)
• trigger_rule - 我們在這個步驟一開始時曾簡短提及觸發規則，但這裡我們會實際示範如何使用。根據預設，除非所有上游運算子均成功完成，否則 Airflow 運算子不會執行。ALL_DONE 觸發規則只要求所有上游運算子都已完成，不論是否成功都一樣。這表示即使 Hadoop 工作失敗，我們仍然想要拆解叢集。

  create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

最後，我們希望這些運算子依特定順序執行，而使用 Python Bitshift 運算子即可表示這點。在此情況下，系統會一律先執行 create_dataproc_cluster，再執行 run_dataproc_hadoop，最後再執行 delete_dataproc_cluster。

將所有程式碼整合在一起，程式碼看起來會像這樣：

# Copyright 2018 Google LLC
#
# 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
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# [START composer_hadoop_tutorial]
"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop
wordcount example, and deletes the cluster.

This DAG relies on three Airflow variables
https://airflow.apache.org/concepts.html#variables
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be
  created.
* gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
  See https://cloud.google.com/storage/docs/creating-buckets for creating a
  bucket.
"""

import datetime
import os

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

# Output file for Cloud Dataproc job.
output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
# Path to Hadoop wordcount example available on every Dataproc cluster.
WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)
# Arguments to pass to Cloud Dataproc job.
input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date': yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

# [START composer_hadoop_schedule]
with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    # [END composer_hadoop_schedule]

    # Create a Cloud Dataproc cluster.
    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/code.html#default-variables
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

    # Delete Cloud Dataproc cluster.
    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
        # even if the Dataproc job fails.
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

    # [START composer_hadoop_steps]
    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster
    # [END composer_hadoop_steps]

# [END composer_hadoop]

如果因為環境並未建立而略過步驟 3，請務必立即返回設定再繼續執行步驟 5。

5. 將 Airflow 檔案上傳至 Cloud Storage

將 DAG 複製到 /dags 資料夾

1. 首先，請開啟 Cloud Shell，該工具會為您方便地安裝 Cloud SDK。
2. 複製 Python 範例存放區，並變更為 Composer/workflows 目錄

git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git && cd python-docs-samples/composer/workflows

3. 執行下列指令，將 DAG 資料夾名稱設為環境變數

DAGS_FOLDER=$(gcloud composer environments describe ${COMPOSER_INSTANCE} \
--location us-central1 --format="value(config.dagGcsPrefix)")

4. 執行下列 gsutil 指令，將教學課程程式碼複製到建立 /dags 資料夾的位置

gsutil cp hadoop_tutorial.py $DAGS_FOLDER

輸出內容應如下所示：

Copying file://hadoop_tutorial.py [Content-Type=text/x-python]...
/ [1 files][  4.1 KiB/  4.1 KiB]
Operation completed over 1 objects/4.1 KiB.

注意：本指南現在將工作流程稱為 composer_hadoop_tutorial

6. 使用 Airflow UI

如要使用 GCP 主控台存取 Airflow 網頁介面，請按照下列步驟操作： 開啟「環境」頁面。 在環境的「Airflow 網路伺服器」欄位，按一下新視窗圖示。Airflow 網頁版 UI 會在新的瀏覽器視窗中開啟。

如要瞭解 Airflow UI，請參閱存取網頁介面。

檢視變數

您先前設定的變數會保留在環境中。您可以從 Airflow UI 選單列中依序選取「管理」>「變數」，查看變數。

探索 DAG 執行作業

當您將 DAG 檔案上傳到 Cloud Storage 中的 dags 資料夾後，Cloud Composer 會剖析檔案。如果找不到錯誤，工作流程名稱就會顯示在 DAG 清單內，且工作流程會排入佇列以立即執行。如要查看 DAG，請按一下頁面頂端的「DAG」。

按一下 composer_hadoop_tutorial 開啟 DAG 詳細資料頁面。此頁面會透過圖形呈現工作流程工作和依附元件。

接著在工具列中點選「圖表檢視」，然後將滑鼠游標懸停在各工作的圖表上，即可查看工作狀態。請注意，每個工作周圍的框線也會指出狀態，例如綠色框線表示執行中，紅色框線表示執行失敗等。

如要再次從「圖表檢視畫面」執行工作流程：

1. 在 Airflow 使用者介面的圖表檢視畫面中，按一下 create_dataproc_cluster 圖示。
2. 按一下「清除」即可重設三項工作，然後按一下「確定」確認。

您也可以前往以下 GCP 控制台頁面，查看「composer-hadoop-tutorial」工作流程的狀態和結果：

• Cloud Dataproc 叢集：監控叢集的建立和刪除作業。請注意，工作流程建立的叢集是暫時性的，只會在工作流程執行期間存在，並在最後一個工作流程工作結束時刪除。
• Cloud Dataproc 工作，查看或監控 Apache Hadoop 字數工作。按一下工作 ID 即可查看工作記錄輸出。
• Cloud Storage 瀏覽器：前往為這個程式碼研究室建立的 Cloud Storage 值區，查看 wordcount 資料夾中的字詞計數結果。

7. 清除

如要避免系統向您的 GCP 帳戶收取您在本程式碼研究室中所用資源的費用，請按照下列步驟操作：

1. (選用) 如要儲存資料，請從 Cloud Storage 值區下載資料，該值區是為 Cloud Composer 環境和您為本程式碼研究室建立的儲存值區。
2. 刪除您為本程式碼研究室建立的 Cloud Storage 值區。
3. 刪除環境的 Cloud Storage 值區。
4. 刪除 Cloud Composer 環境。請注意，刪除環境並不會刪除環境的儲存空間值區。

您也可以選擇刪除專案：

1. 前往 GCP 主控台的「Projects」頁面。
2. 在專案清單中選取要刪除的專案，然後按一下「刪除」。
3. 在方塊中輸入專案 ID，然後按一下「Shut down」(關閉) 即可刪除專案。

注意：刪除專案會出現以下結果：

• 專案中的所有內容都會遭到刪除。 若您使用現有專案進行本教學課程，當您刪除專案時，也會一併刪除您在該專案中完成的任何其他工作。
• 自訂專案 ID 不復存在。在您建立這個專案時，可能採用日後要使用的自訂專案 ID。如要保留使用該專案 ID 的網址 (例如 appspot.com 網址)，請刪除在該專案中選取的資源，而不是刪除整個專案。