Google Cloud Memorystore 運算子¶
Cloud Memorystore for Redis 是 Google Cloud 的全受管 Redis 服務。在 Google Cloud 上執行的應用程式可以利用高度可擴展、可用、安全的 Redis 服務來實現極致效能，而無需承擔管理複雜 Redis 部署的負擔。
先決條件任務¶
若要使用這些運算子，您必須完成幾件事
使用 Cloud Console 選取或建立 Cloud Platform 專案。
為您的專案啟用計費功能，如 Google Cloud 文件 中所述。
啟用 API，如 Cloud Console 文件 中所述。
透過 pip 安裝 API 函式庫。
pip install 'apache-airflow[google]'如需詳細資訊，請參閱 安裝。
執行個體¶
運算子使用 Instance 來表示執行個體。該物件也可以表示為相容的字典。
以下是一個執行個體的範例
FIRST_INSTANCE = {"tier": Instance.Tier.BASIC, "memory_size_gb": 1}
設定匯入/匯出儲存桶權限¶
必須設定儲存桶的權限才能匯入和匯出資料。若要尋找執行個體的服務帳戶，請執行 CloudMemorystoreCreateInstanceOperator 或 CloudMemorystoreGetInstanceOperator，並使用在 persistenceIamIdentity 下列出的服務帳戶。
您可以使用 GCSBucketCreateAclEntryOperator 運算子來設定權限。
set_acl_permission = GCSBucketCreateAclEntryOperator(
task_id="gcs-set-acl-permission",
bucket=BUCKET_NAME,
entity="user-{{ task_instance.xcom_pull('get-instance')['persistence_iam_identity']"
".split(':', 2)[1] }}",
role="OWNER",
)
如需更多資訊，請參閱：授予匯入和匯出的限制權限
建立執行個體¶
建立執行個體是透過 CloudMemorystoreCreateInstanceOperator 運算子執行。
create_instance = CloudMemorystoreCreateInstanceOperator(
task_id="create-instance",
location=LOCATION,
instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME,
instance=FIRST_INSTANCE,
project_id=PROJECT_ID,
)
您可以搭配 location、 instance_id、 instance、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。結果會儲存到 XCom，以便其他運算子使用。
create_instance_result = BashOperator(
task_id="create-instance-result",
bash_command=f"echo {create_instance.output}",
)
刪除執行個體¶
刪除執行個體是透過 CloudMemorystoreDeleteInstanceOperator 運算子執行。
delete_instance = CloudMemorystoreDeleteInstanceOperator(
task_id="delete-instance",
location=LOCATION,
instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
project_id=PROJECT_ID,
)
您可以搭配 location、 instance、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
匯出執行個體¶
刪除執行個體是透過 CloudMemorystoreExportInstanceOperator 運算子執行。
export_instance = CloudMemorystoreExportInstanceOperator(
task_id="export-instance",
location=LOCATION,
instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
output_config={"gcs_destination": {"uri": EXPORT_GCS_URL}},
project_id=PROJECT_ID,
)
您可以搭配 location、 instance、 output_config、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
容錯移轉執行個體¶
刪除執行個體是透過 CloudMemorystoreFailoverInstanceOperator 運算子執行。
failover_instance = CloudMemorystoreFailoverInstanceOperator(
task_id="failover-instance",
location=LOCATION,
instance=MEMORYSTORE_REDIS_INSTANCE_NAME_2,
data_protection_mode=FailoverInstanceRequest.DataProtectionMode(
FailoverInstanceRequest.DataProtectionMode.LIMITED_DATA_LOSS
),
project_id=PROJECT_ID,
)
您可以搭配 location、 instance、 data_protection_mode、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
取得執行個體¶
刪除執行個體是透過 CloudMemorystoreGetInstanceOperator 運算子執行。
get_instance = CloudMemorystoreGetInstanceOperator(
task_id="get-instance",
location=LOCATION,
instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
project_id=PROJECT_ID,
do_xcom_push=True,
)
您可以搭配 location、 instance、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
匯入執行個體¶
刪除執行個體是透過 CloudMemorystoreImportOperator 運算子執行。
import_instance = CloudMemorystoreImportOperator(
task_id="import-instance",
location=LOCATION,
instance=MEMORYSTORE_REDIS_INSTANCE_NAME_2,
input_config={"gcs_source": {"uri": EXPORT_GCS_URL}},
project_id=PROJECT_ID,
)
您可以搭配 location、 instance、 input_config、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
列出執行個體¶
列出執行個體是透過 CloudMemorystoreListInstancesOperator 運算子執行。
list_instances = CloudMemorystoreListInstancesOperator(
task_id="list-instances", location="-", page_size=100, project_id=PROJECT_ID
)
您可以搭配 location、 page_size、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。結果會儲存到 XCom，以便其他運算子使用。
list_instances_result = BashOperator(
task_id="list-instances-result", bash_command=f"echo {get_instance.output}"
)
更新執行個體¶
更新執行個體是透過 CloudMemorystoreUpdateInstanceOperator 運算子執行。
update_instance = CloudMemorystoreUpdateInstanceOperator(
task_id="update-instance",
location=LOCATION,
instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME,
project_id=PROJECT_ID,
update_mask={"paths": ["memory_size_gb"]},
instance={"memory_size_gb": 2},
)
您可以搭配 update_mask、 instance、 location、 instance_id、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
調整執行個體規模¶
調整執行個體規模是透過 CloudMemorystoreScaleInstanceOperator 運算子執行。
scale_instance = CloudMemorystoreScaleInstanceOperator(
task_id="scale-instance",
location=LOCATION,
instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME_3,
project_id=PROJECT_ID,
memory_size_gb=3,
)
您可以搭配 memory_size_gb、 location、 instance_id、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。
建立執行個體並匯入¶
如果您想要建立具有匯入資料的執行個體，可以使用 CloudMemorystoreCreateInstanceAndImportOperator 運算子。
匯出並刪除執行個體¶
如果您想要匯出資料並立即刪除執行個體，則可以使用 CloudMemorystoreExportAndDeleteInstanceOperator 運算子。
您可以搭配 memory_size_gb、 location、 instance_id、 project_id、 retry、 timeout、 metadata、 gcp_conn_id、 impersonation_chain 參數使用 Jinja 範本，讓您可以動態決定值。