תבנית להעברה מ-Apache Cassandra ל-Bigtable

התבנית Apache Cassandra to Bigtable מעתיקה טבלה מ-Apache Cassandra ל-Bigtable. התבנית הזו דורשת הגדרה מינימלית, והיא משכפלת את מבנה הטבלה ב-Cassandra בצורה הכי מדויקת שאפשר ב-Bigtable.

התבנית Apache Cassandra to Bigtable שימושית למטרות הבאות:

• העברת מסד נתונים של Apache Cassandra כשזמן השבתה קצר הוא סביר.
• שכפול תקופתי של טבלאות Cassandra ל-Bigtable לצורך הגשה גלובלית.

הדרישות לגבי צינורות עיבוד נתונים

• טבלת היעד ב-Bigtable צריכה להתקיים לפני שמריצים את צינור הנתונים.
• חיבור רשת בין עובדי Dataflow לבין צמתי Apache Cassandra.

המרת סוג

תבנית Apache Cassandra to Bigtable ממירה באופן אוטומטי סוגי נתונים של Apache Cassandra לסוגי נתונים של Bigtable.

רוב הפרימיטיבים מיוצגים באותו אופן ב-Bigtable וב-Apache Cassandra, אבל הפרימיטיבים הבאים מיוצגים באופן שונה:

• ‫Date ו-Timestamp מומרים לאובייקטים של DateTime
• ההודעה UUID הומרה ל-String
• ההודעה Varint הומרה ל-BigDecimal

‫Apache Cassandra תומך באופן מובנה גם בסוגים מורכבים יותר כמו Tuple, ‏ List, ‏ Set ו-Map. הצינור הזה לא תומך בטופלים כי אין סוג תואם ב-Apache Beam.

לדוגמה, ב-Apache Cassandra יכולה להיות עמודה מסוג List בשם mylist עם ערכים כמו אלה שבטבלה הבאה:

שורה	mylist
1	(a,b,c)

הצינור מרחיב את עמודת הרשימה לשלוש עמודות שונות (שנקראות ב-Bigtable מזהים של עמודות). שם העמודה הוא mylist, אבל הצינור מוסיף את האינדקס של הפריט ברשימה, למשל mylist[0].

שורה	mylist[0]	mylist[1]	mylist[2]
1	a	b	c

הצינור מטפל בקבוצות באותו אופן שבו הוא מטפל ברשימות, אבל הוא מוסיף סיומת כדי לציין אם התא הוא מפתח או ערך.

שורה	mymap
1	{"first_key":"first_value","another_key":"different_value"}

אחרי השינוי, הטבלה תיראה כך:

שורה	mymap[0].key	mymap[0].value	mymap[1].key	mymap[1].value
1	first_key	first_value	another_key	different_value

המרת מפתח ראשי

ב-Apache Cassandra, מפתח ראשי מוגדר באמצעות שפת הגדרת נתונים. המפתח הראשי הוא פשוט, מורכב או משולב עם עמודות האשכול. ‫Bigtable תומך ביצירה ידנית של מפתחות שורות, שמסודרים בסדר לקסיקוגרפי במערך בייטים. תהליך הצינור אוסף באופן אוטומטי מידע על סוג המפתח ובונה מפתח על סמך שיטות מומלצות לבניית מפתחות שורות שמבוססים על כמה ערכים.

פרמטרים של תבניות

פרמטרים נדרשים

• ‫cassandraHosts: רשימה מופרדת בפסיקים של המארחים של צמתי Apache Cassandra.
• ‫cassandraKeyspace: מרחב המפתחות של Apache Cassandra שבו נמצא הטבלה.
• ‫cassandraTable: טבלת Apache Cassandra להעתקה.
• ‫bigtableProjectId: מזהה הפרויקט ב-Google Cloud שמשויך למופע Bigtable.
• ‫bigtableInstanceId: המזהה של מכונת Bigtable שאליה מועתקת טבלת Apache Cassandra.
• ‫bigtableTableId: השם של טבלת Bigtable שאליה מועתקת טבלת Apache Cassandra.

פרמטרים אופציונליים

• ‫cassandraPort: יציאת ה-TCP שמשמשת להגעה ל-Apache Cassandra בצמתים. ערך ברירת המחדל הוא 9042.
• ‫defaultColumnFamily: השם של קבוצת העמודות בטבלת Bigtable. ערך ברירת המחדל הוא default.
• ‫rowKeySeparator: המפריד שמשמש ליצירת מפתחות שורות. ערך ברירת המחדל הוא #.
• ‫splitLargeRows: הדגל להפעלת פיצול של שורות גדולות לכמה בקשות MutateRows. חשוב לזכור שכאשר שורה גדולה מפולחת בין כמה קריאות API, העדכונים לשורה לא מתבצעים באופן אטומי. .
• ‫writetimeCassandraColumnSchema: נתיב GCS לסכימה להעתקת נתוני writetime של Cassandra ל-Bigtable. הפקודה ליצירת הסכימה הזו היא cqlsh -e "select json * from system_schema.columns where keyspace_name='$CASSANDRA_KEYSPACE' and table_name='$CASSANDRA_TABLE'`" > column_schema.json. מגדירים את $WRITETIME_CASSANDRA_COLUMN_SCHEMA לנתיב GCS, לדוגמה gs://$BUCKET_NAME/column_schema.json. לאחר מכן מעלים את הסכימה ל-GCS: gcloud storage cp column_schema.json $WRITETIME_CASSANDRA_COLUMN_SCHEMA. נדרשת גרסה 2.2 ואילך של Cassandra לתמיכה ב-JSON.
• ‫setZeroTimestamp: הדגל להגדרת חותמת הזמן של תא Bigtable ל-0 אם לא מופיע writetime של Cassandra. ההתנהגות שמוגדרת כברירת מחדל כשלא מגדירים את הדגל הזה היא להגדיר את חותמת הזמן של תא Bigtable כזמן השכפול של התבנית, כלומר עכשיו.

הפעלת התבנית

המסוף

1. עוברים לדף Dataflow Create job from template (יצירת משימה מתבנית).
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Cassandra to Cloud Bigtable template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות classic, צריך להתקין את Google Cloud CLI בגרסה 138.0.0, או בגרסה מאוחרת יותר.

במעטפת או בטרמינל, מריצים את התבנית:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/Cassandra_To_Cloud_Bigtable \
    --region REGION_NAME \
    --parameters \
bigtableProjectId=BIGTABLE_PROJECT_ID,\
bigtableInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableTableId=BIGTABLE_TABLE_ID,\
cassandraHosts=CASSANDRA_HOSTS,\
cassandraKeyspace=CASSANDRA_KEYSPACE,\
cassandraTable=CASSANDRA_TABLE

מחליפים את מה שכתוב בשדות הבאים:

• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫BIGTABLE_PROJECT_ID: מזהה הפרויקט שבו נמצא Bigtable
• ‫BIGTABLE_INSTANCE_ID: מזהה מכונת Bigtable
• ‫BIGTABLE_TABLE_ID: השם של טבלת Bigtable
• ‫CASSANDRA_HOSTS: רשימת המארחים של Apache Cassandra. אם מספקים כמה מארחים, צריך לפעול לפי ההוראות להוספת פסיקים
• ‫CASSANDRA_KEYSPACE: מרחב המפתחות של Apache Cassandra שבו נמצאת הטבלה
• ‫CASSANDRA_TABLE: טבלת Apache Cassandra שצריך להעביר

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/Cassandra_To_Cloud_Bigtable
{
   "jobName": "JOB_NAME",
   "parameters": {
       "bigtableProjectId": "BIGTABLE_PROJECT_ID",
       "bigtableInstanceId": "BIGTABLE_INSTANCE_ID",
       "bigtableTableId": "BIGTABLE_TABLE_ID",
       "cassandraHosts": "CASSANDRA_HOSTS",
       "cassandraKeyspace": "CASSANDRA_KEYSPACE",
       "cassandraTable": "CASSANDRA_TABLE"
   },
   "environment": { "zone": "us-central1-f" }
}

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫BIGTABLE_PROJECT_ID: מזהה הפרויקט שבו נמצא Bigtable
• ‫BIGTABLE_INSTANCE_ID: מזהה מכונת Bigtable
• ‫BIGTABLE_TABLE_ID: השם של טבלת Bigtable
• ‫CASSANDRA_HOSTS: רשימת המארחים של Apache Cassandra. אם מספקים כמה מארחים, צריך לפעול לפי ההוראות להוספת פסיקים
• ‫CASSANDRA_KEYSPACE: מרחב המפתחות של Apache Cassandra שבו נמצאת הטבלה
• ‫CASSANDRA_TABLE: טבלת Apache Cassandra שצריך להעביר

קוד המקור של התבנית

Java

/*
 * Copyright (C) 2019 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
 *
 *   http://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.
 */
package com.google.cloud.teleport.bigtable;

import com.datastax.driver.core.Session;
import com.google.cloud.teleport.bigtable.CassandraToBigtable.Options;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.cassandra.CassandraIO;
import org.apache.beam.sdk.io.cassandra.Mapper;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This Dataflow Template performs a one off copy of one table from Apache Cassandra to Cloud
 * Bigtable. It is designed to require minimal configuration and aims to replicate the table
 * structure in Cassandra as closely as possible in Cloud Bigtable.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_Cassandra_To_Cloud_Bigtable.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Cassandra_To_Cloud_Bigtable",
    category = TemplateCategory.BATCH,
    displayName = "Cassandra to Cloud Bigtable",
    description = {
      "The Apache Cassandra to Cloud Bigtable template copies a table from Apache Cassandra to Cloud Bigtable. "
          + "This template requires minimal configuration and replicates the table structure in Cassandra as closely as possible in Cloud Bigtable.",
      "The Apache Cassandra to Cloud Bigtable template is useful for the following:\n"
          + "- Migrating Apache Cassandra database when short downtime is acceptable.\n"
          + "- Periodically replicating Cassandra tables to Cloud Bigtable for global serving."
    },
    optionsClass = Options.class,
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cassandra-to-bigtable",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The target Bigtable table must exist before running the pipeline.",
      "Network connection between Dataflow workers and Apache Cassandra nodes."
    })
public final class CassandraToBigtable {

  /** TODO - refactor to extend BigtableCommonOptions.WriteOptions. */
  public interface Options extends PipelineOptions {

    @TemplateParameter.Text(
        order = 1,
        groupName = "Source",
        regexes = {"^[a-zA-Z0-9\\.\\-,]*$"},
        description = "Cassandra Hosts",
        helpText = "The hosts of the Apache Cassandra nodes in a comma-separated list.")
    ValueProvider<String> getCassandraHosts();

    @SuppressWarnings("unused")
    void setCassandraHosts(ValueProvider<String> hosts);

    @TemplateParameter.Integer(
        order = 2,
        groupName = "Source",
        optional = true,
        description = "Cassandra Port",
        helpText =
            "The TCP port to use to reach Apache Cassandra on the nodes. The default value is `9042`.")
    @Default.Integer(9042)
    ValueProvider<Integer> getCassandraPort();

    @SuppressWarnings("unused")
    void setCassandraPort(ValueProvider<Integer> port);

    @TemplateParameter.Text(
        order = 3,
        groupName = "Source",
        regexes = {"^[a-zA-Z0-9][a-zA-Z0-9_]{0,47}$"},
        description = "Cassandra Keyspace",
        helpText = "The Apache Cassandra keyspace where the table is located.")
    ValueProvider<String> getCassandraKeyspace();

    @SuppressWarnings("unused")
    void setCassandraKeyspace(ValueProvider<String> keyspace);

    @TemplateParameter.Text(
        order = 4,
        groupName = "Source",
        regexes = {"^[a-zA-Z][a-zA-Z0-9_]*$"},
        description = "Cassandra Table",
        helpText = "The Apache Cassandra table to copy.")
    ValueProvider<String> getCassandraTable();

    @SuppressWarnings("unused")
    void setCassandraTable(ValueProvider<String> cassandraTable);

    @TemplateParameter.ProjectId(
        order = 5,
        groupName = "Target",
        description = "Bigtable Project ID",
        helpText = "The Google Cloud project ID associated with the Bigtable instance.")
    ValueProvider<String> getBigtableProjectId();

    @SuppressWarnings("unused")
    void setBigtableProjectId(ValueProvider<String> projectId);

    @TemplateParameter.Text(
        order = 6,
        groupName = "Target",
        regexes = {"[a-z][a-z0-9\\-]+[a-z0-9]"},
        description = "Target Bigtable Instance",
        helpText = "The ID of the Bigtable instance that the Apache Cassandra table is copied to.")
    ValueProvider<String> getBigtableInstanceId();

    @SuppressWarnings("unused")
    void setBigtableInstanceId(ValueProvider<String> bigtableInstanceId);

    @TemplateParameter.Text(
        order = 7,
        groupName = "Target",
        regexes = {"[_a-zA-Z0-9][-_.a-zA-Z0-9]*"},
        description = "Target Bigtable Table",
        helpText = "The name of the Bigtable table that the Apache Cassandra table is copied to.")
    ValueProvider<String> getBigtableTableId();

    @SuppressWarnings("unused")
    void setBigtableTableId(ValueProvider<String> bigtableTableId);

    @TemplateParameter.Text(
        order = 8,
        groupName = "Target",
        optional = true,
        regexes = {"[-_.a-zA-Z0-9]+"},
        description = "The Default Bigtable Column Family",
        helpText =
            "The name of the column family of the Bigtable table. The default value is `default`.")
    @Default.String("default")
    ValueProvider<String> getDefaultColumnFamily();

    @SuppressWarnings("unused")
    void setDefaultColumnFamily(ValueProvider<String> defaultColumnFamily);

    @TemplateParameter.Text(
        order = 9,
        groupName = "Target",
        optional = true,
        description = "The Row Key Separator",
        helpText = "The separator used to build row-keys. The default value is `#`.")
    @Default.String("#")
    ValueProvider<String> getRowKeySeparator();

    @SuppressWarnings("unused")
    void setRowKeySeparator(ValueProvider<String> rowKeySeparator);

    @TemplateParameter.Boolean(
        order = 10,
        groupName = "Target",
        optional = true,
        description = "If true, large rows will be split into multiple MutateRows requests",
        helpText =
            "The flag for enabling splitting of large rows into multiple MutateRows requests. Note that when a large row is split between multiple API calls, the updates to the row are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);

    // TODO(georgecma) - upgrade template to V2 or modify CassandraIO so column schema is
    // automatically processed.
    @TemplateParameter.GcsReadFile(
        order = 11,
        optional = true,
        description =
            "GCS path to Cassandra JSON column schema. If set, the template will use the schema info to copy Cassandra write time to Bigtable cells",
        helpText =
            "GCS path to schema to copy Cassandra writetimes to Bigtable. The command to generate this schema is ```cqlsh -e \"select json * from system_schema.columns where keyspace_name='$CASSANDRA_KEYSPACE' and table_name='$CASSANDRA_TABLE'`\" > column_schema.json```. Set $WRITETIME_CASSANDRA_COLUMN_SCHEMA to a GCS path, e.g. `gs://$BUCKET_NAME/column_schema.json`. Then upload the schema to GCS: `gcloud storage cp column_schema.json $WRITETIME_CASSANDRA_COLUMN_SCHEMA`. Requires Cassandra version 2.2 onwards for JSON support.")
    ValueProvider<String> getWritetimeCassandraColumnSchema();

    void setWritetimeCassandraColumnSchema(ValueProvider<String> writetimeCassandraColumnSchema);

    @TemplateParameter.Boolean(
        order = 12,
        optional = true,
        description =
            "If true, sets Bigtable cell timestamp to 0 if writetime is not present. Else, the Bigtable cell timestamp defaults to pipeline run time (i.e. now).",
        helpText =
            "The flag for setting Bigtable cell timestamp to 0 if Cassandra writetime is not present. The default behavior for when this flag is not set is to set the Bigtable cell timestamp as the template replication time, i.e. now.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getSetZeroTimestamp();

    void setSetZeroTimestamp(ValueProvider<Boolean> setZeroTimestamp);
  }

  private static final Logger LOG = LoggerFactory.getLogger(CassandraToBigtable.class);

  /**
   * Runs a pipeline to copy one Cassandra table to Cloud Bigtable.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) throws Exception {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    // Split the Cassandra Hosts value provider into a list value provider.
    ValueProvider.NestedValueProvider<List<String>, String> hosts =
        ValueProvider.NestedValueProvider.of(
            options.getCassandraHosts(),
            (SerializableFunction<String, List<String>>) value -> Arrays.asList(value.split(",")));

    Pipeline p = Pipeline.create(PipelineUtils.tweakPipelineOptions(options));

    // Create a factory method to inject the CassandraRowMapperFn to allow custom type mapping.
    SerializableFunction<Session, Mapper> cassandraObjectMapperFactory =
        new CassandraRowMapperFactory(options.getCassandraTable(), options.getCassandraKeyspace());

    CassandraIO.Read<Row> source =
        CassandraIO.<Row>read()
            .withHosts(hosts)
            .withPort(options.getCassandraPort())
            .withKeyspace(options.getCassandraKeyspace())
            .withTable(options.getCassandraTable())
            .withMapperFactoryFn(cassandraObjectMapperFactory)
            .withEntity(Row.class)
            .withCoder(SerializableCoder.of(Row.class))
            .withQuery(
                new CassandraWritetimeQueryProvider(
                    options.getWritetimeCassandraColumnSchema(),
                    options.getCassandraKeyspace(),
                    options.getCassandraTable()));

    BigtableIO.Write sink =
        BigtableIO.write()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

    p.apply("Read from Cassandra", source)
        .apply(
            "Convert Row",
            ParDo.of(
                BeamRowToBigtableFn.createWithSplitLargeRows(
                    options.getRowKeySeparator(),
                    options.getDefaultColumnFamily(),
                    options.getSplitLargeRows(),
                    BeamRowToBigtableFn.MAX_MUTATION_PER_REQUEST,
                    options.getWritetimeCassandraColumnSchema(),
                    options.getSetZeroTimestamp())))
        .apply("Write to Bigtable", sink);
    p.run();
  }
}

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• רשימת התבניות ש-Google סיפקה