Google Cloud Memorystore Operators

The Cloud Memorystore for Redis is a fully managed Redis service for the Google Cloud. Applications running on Google Cloud can achieve extreme performance by leveraging the highly scalable, available, secure Redis service without the burden of managing complex Redis deployments.

Instance

Operators uses a Instance for representing instance. The object can be presented as a compatible dictionary also.

Here is an example of instance

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

FIRST_INSTANCE = {"tier": Instance.Tier.BASIC, "memory_size_gb": 1}

Configuration of bucket permissions for import / export

It is necessary to configure permissions for the bucket to import and export data. Too find the service account for your instance, run the CloudMemorystoreCreateInstanceOperator or CloudMemorystoreGetInstanceOperator and make a use of the service account listed under persistenceIamIdentity.

You can use GCSBucketCreateAclEntryOperator operator to set permissions.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

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",
)

For further information look at: Granting restricted permissions for import and export

Create instance

Create a instance is performed with the CloudMemorystoreCreateInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

create_instance = CloudMemorystoreCreateInstanceOperator(
    task_id="create-instance",
    location="europe-north1",
    instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME,
    instance=FIRST_INSTANCE,
    project_id=GCP_PROJECT_ID,
)

You can use Jinja templating with location, instance_id, instance, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values. The result is saved to XCom, which allows it to be used by other operators.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

create_instance_result = BashOperator(
    task_id="create-instance-result",
    bash_command="echo \"{{ task_instance.xcom_pull('create-instance') }}\"",
)

Delete instance

Delete a instance is performed with the CloudMemorystoreDeleteInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

delete_instance = CloudMemorystoreDeleteInstanceOperator(
    task_id="delete-instance",
    location="europe-north1",
    instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
    project_id=GCP_PROJECT_ID,
)

You can use Jinja templating with location, instance, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Export instance

Delete a instance is performed with the CloudMemorystoreExportInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

export_instance = CloudMemorystoreExportInstanceOperator(
    task_id="export-instance",
    location="europe-north1",
    instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
    output_config={"gcs_destination": {"uri": EXPORT_GCS_URL}},
    project_id=GCP_PROJECT_ID,
)

You can use Jinja templating with location, instance, output_config, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Failover instance

Delete a instance is performed with the CloudMemorystoreFailoverInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

failover_instance = CloudMemorystoreFailoverInstanceOperator(
    task_id="failover-instance",
    location="europe-north1",
    instance=MEMORYSTORE_REDIS_INSTANCE_NAME_2,
    data_protection_mode=FailoverInstanceRequest.DataProtectionMode.LIMITED_DATA_LOSS,
    project_id=GCP_PROJECT_ID,
)

You can use Jinja templating with location, instance, data_protection_mode, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Get instance

Delete a instance is performed with the CloudMemorystoreGetInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

get_instance = CloudMemorystoreGetInstanceOperator(
    task_id="get-instance",
    location="europe-north1",
    instance=MEMORYSTORE_REDIS_INSTANCE_NAME,
    project_id=GCP_PROJECT_ID,
    do_xcom_push=True,
)

You can use Jinja templating with location, instance, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Import instance

Delete a instance is performed with the CloudMemorystoreImportOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

import_instance = CloudMemorystoreImportOperator(
    task_id="import-instance",
    location="europe-north1",
    instance=MEMORYSTORE_REDIS_INSTANCE_NAME_2,
    input_config={"gcs_source": {"uri": EXPORT_GCS_URL}},
    project_id=GCP_PROJECT_ID,
)

You can use Jinja templating with location, instance, input_config, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

List instances

List a instances is performed with the CloudMemorystoreListInstancesOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

list_instances = CloudMemorystoreListInstancesOperator(
    task_id="list-instances", location="-", page_size=100, project_id=GCP_PROJECT_ID
)

You can use Jinja templating with location, page_size, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values. The result is saved to XCom, which allows it to be used by other operators.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

list_instances_result = BashOperator(
    task_id="list-instances-result", bash_command="echo \"{{ task_instance.xcom_pull('get-instance') }}\""
)

Update instance

Update a instance is performed with the CloudMemorystoreUpdateInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

update_instance = CloudMemorystoreUpdateInstanceOperator(
    task_id="update-instance",
    location="europe-north1",
    instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME,
    project_id=GCP_PROJECT_ID,
    update_mask={"paths": ["memory_size_gb"]},
    instance={"memory_size_gb": 2},
)

You can use Jinja templating with update_mask, instance, location, instance_id, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Scale instance

Scale a instance is performed with the CloudMemorystoreScaleInstanceOperator operator.

airflow/providers/google/cloud/example_dags/example_cloud_memorystore.pyView Source

scale_instance = CloudMemorystoreScaleInstanceOperator(
    task_id="scale-instance",
    location="europe-north1",
    instance_id=MEMORYSTORE_REDIS_INSTANCE_NAME_3,
    project_id=GCP_PROJECT_ID,
    memory_size_gb=3,
)

You can use Jinja templating with memory_size_gb, location, instance_id, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Create instance and import

If you want to create instances with imported data, you can use CloudMemorystoreCreateInstanceAndImportOperator operator.

Export and delete instance

If you want to export data and immediately delete instances then you can use CloudMemorystoreExportAndDeleteInstanceOperator operator.

You can use Jinja templating with memory_size_gb, location, instance_id, project_id, retry, timeout, metadata, gcp_conn_id, impersonation_chain parameters which allows you to dynamically determine values.

Was this entry helpful?