Mengimpor, mengekspor, dan mengubah data menggunakan Dataflow

Halaman ini menjelaskan cara menggunakan konektor Dataflow untuk Spanner guna mengimpor, mengekspor, dan mengubah data di database logika Google SQL Spanner dan database logika PostgreSQL.

Dataflow adalah layanan terkelola untuk mengubah dan memperkaya data. Konektor Dataflow untuk Spanner memungkinkan Anda membaca data dari dan menulis data ke Spanner dalam pipeline Dataflow, dengan opsi mengubah atau memodifikasi data. Anda juga dapat membuat pipeline yang mentransfer data antara Spanner dan produk Google Cloud lainnya.

Konektor Dataflow adalah metode yang direkomendasikan untuk memindahkan data secara massal ke dalam dan ke luar Spanner secara efisien. Ini juga merupakan metode yang direkomendasikan untuk melakukan transformasi besar ke database yang tidak didukung oleh DML Berpartisi, seperti pemindahan tabel dan penghapusan massal yang memerlukan JOIN. Saat menggunakan database individual, ada metode lain yang dapat Anda gunakan untuk mengimpor dan mengekspor data:

  • Gunakan konsol Google Cloud untuk mengekspor setiap database dari Spanner ke Cloud Storage dalam format Avro.
  • Gunakan konsol Google Cloud untuk mengimpor database kembali ke Spanner dari file yang Anda ekspor ke Cloud Storage.
  • Gunakan REST API atau Google Cloud CLI untuk menjalankan tugas ekspor atau impor dari Spanner ke Cloud Storage dan sebaliknya juga menggunakan format Avro.

Konektor Dataflow untuk Spanner adalah bagian dari Apache Beam Java SDK, dan menyediakan API untuk melakukan tindakan sebelumnya. Untuk informasi selengkapnya tentang beberapa konsep yang dibahas di halaman ini, seperti objek dan transformasi PCollection, lihat panduan pemrograman Apache Beam.

Menambahkan konektor ke project Maven Anda

Untuk menambahkan konektor Google Cloud Dataflow ke project Maven, tambahkan artefak Maven beam-sdks-java-io-google-cloud-platform ke file pom.xml sebagai dependensi.

Misalnya, dengan asumsi bahwa file pom.xml menetapkan beam.version ke nomor versi yang sesuai, Anda akan menambahkan dependensi berikut:

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>${beam.version}</version>
</dependency>

Membaca data dari Spanner

Untuk membaca dari Spanner, terapkan transformasi SpannerIO.read. Konfigurasikan pembacaan menggunakan metode di class SpannerIO.Read. Menerapkan transformasi akan menampilkan PCollection<Struct>, dengan setiap elemen dalam koleksi mewakili setiap baris yang ditampilkan oleh operasi baca. Anda dapat membaca dari Spanner dengan dan tanpa kueri SQL tertentu, bergantung pada output yang diperlukan.

Menerapkan transformasi SpannerIO.read akan menampilkan tampilan data yang konsisten dengan melakukan pembacaan yang kuat. Kecuali jika Anda menentukan lain, hasil pembacaan akan diambil snapshot-nya pada saat Anda memulai pembacaan. Lihat pembacaan untuk mengetahui informasi selengkapnya tentang berbagai jenis pembacaan yang dapat dilakukan Spanner.

Membaca data menggunakan kueri

Untuk membaca kumpulan data tertentu dari Spanner, konfigurasikan transformasi menggunakan metode SpannerIO.Read.withQuery untuk menentukan kueri SQL. Contoh:

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withQuery("SELECT * FROM " + options.getTable()));

Membaca data tanpa menentukan kueri

Untuk membaca dari database tanpa menggunakan kueri, Anda dapat menentukan nama tabel menggunakan metode SpannerIO.Read.withTable, dan menentukan daftar kolom yang akan dibaca menggunakan metode SpannerIO.Read.withColumns. Contoh:

GoogleSQL 

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withTable("Singers")
        .withColumns("singerId", "firstName", "lastName"));

PostgreSQL 

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withTable("singers")
        .withColumns("singer_id", "first_name", "last_name"));

Untuk membatasi baris yang dibaca, Anda dapat menentukan kumpulan kunci utama yang akan dibaca menggunakan metode SpannerIO.Read.withKeySet.

Anda juga dapat membaca tabel menggunakan indeks sekunder yang ditentukan. Seperti halnya panggilan readUsingIndex API, indeks harus berisi semua data yang muncul dalam hasil kueri.

Untuk melakukannya, tentukan tabel seperti yang ditunjukkan pada contoh sebelumnya, dan tentukan indeks yang berisi nilai kolom yang diperlukan menggunakan metode SpannerIO.Read.withIndex. Indeks harus menyimpan semua kolom yang perlu dibaca transformasi. Kunci utama tabel dasar disimpan secara implisit. Misalnya, untuk membaca tabel Songs menggunakan indeks SongsBySongName, Anda menggunakan kode berikut:

GoogleSQL 

// Read the indexed columns from all rows in the specified index.
PCollection<Struct> records =
    pipeline.apply(
        SpannerIO.read()
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId)
            .withTable("Songs")
            .withIndex("SongsBySongName")
            // Can only read columns that are either indexed, STORED in the index or
            // part of the primary key of the Songs table,
            .withColumns("SingerId", "AlbumId", "TrackId", "SongName"));

PostgreSQL 

// // Read the indexed columns from all rows in the specified index.
PCollection<Struct> records =
    pipeline.apply(
        SpannerIO.read()
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId)
            .withTable("Songs")
            .withIndex("SongsBySongName")
            // Can only read columns that are either indexed, STORED in the index or
            // part of the primary key of the songs table,
            .withColumns("singer_id", "album_id", "track_id", "song_name"));

Mengontrol keusangan data transaksi

Transformasi dijamin akan dieksekusi pada snapshot data yang konsisten. Untuk mengontrol ketidak-berlakuan data, gunakan metode SpannerIO.Read.withTimestampBound. Lihat transaksi untuk mengetahui informasi selengkapnya.

Membaca dari beberapa tabel dalam transaksi yang sama

Jika Anda ingin membaca data dari beberapa tabel pada titik waktu yang sama untuk memastikan konsistensi data, lakukan semua operasi baca dalam satu transaksi. Untuk melakukannya, terapkan transformasi createTransaction, yang membuat objek PCollectionView<Transaction> yang kemudian membuat transaksi. Tampilan yang dihasilkan dapat diteruskan ke operasi baca menggunakan SpannerIO.Read.withTransaction.

GoogleSQL 

SpannerConfig spannerConfig =
    SpannerConfig.create().withInstanceId(instanceId).withDatabaseId(databaseId);
PCollectionView<Transaction> tx =
    pipeline.apply(
        SpannerIO.createTransaction()
            .withSpannerConfig(spannerConfig)
            .withTimestampBound(TimestampBound.strong()));
PCollection<Struct> singers =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT SingerID, FirstName, LastName FROM Singers")
            .withTransaction(tx));
PCollection<Struct> albums =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT SingerId, AlbumId, AlbumTitle FROM Albums")
            .withTransaction(tx));

PostgreSQL 

SpannerConfig spannerConfig =
    SpannerConfig.create().withInstanceId(instanceId).withDatabaseId(databaseId);
PCollectionView<Transaction> tx =
    pipeline.apply(
        SpannerIO.createTransaction()
            .withSpannerConfig(spannerConfig)
            .withTimestampBound(TimestampBound.strong()));
PCollection<Struct> singers =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT singer_id, first_name, last_name FROM singers")
            .withTransaction(tx));
PCollection<Struct> albums =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT singer_id, album_id, album_title FROM albums")
            .withTransaction(tx));

Membaca data dari semua tabel yang tersedia

Anda dapat membaca data dari semua tabel yang tersedia di database Spanner.

GoogleSQL 

PCollection<Struct> allRecords =
    pipeline
        .apply(
            SpannerIO.read()
                .withSpannerConfig(spannerConfig)
                .withBatching(false)
                .withQuery(
                    "SELECT t.table_name FROM information_schema.tables AS t WHERE t"
                        + ".table_catalog = '' AND t.table_schema = ''"))
        .apply(
            MapElements.into(TypeDescriptor.of(ReadOperation.class))
                .via(
                    (SerializableFunction<Struct, ReadOperation>)
                        input -> {
                          String tableName = input.getString(0);
                          return ReadOperation.create().withQuery("SELECT * FROM " + tableName);
                        }))
        .apply(SpannerIO.readAll().withSpannerConfig(spannerConfig));

PostgreSQL 

PCollection<Struct> allRecords =
    pipeline
        .apply(
            SpannerIO.read()
                .withSpannerConfig(spannerConfig)
                .withBatching(false)
                .withQuery(
                    Statement.newBuilder(
                            "SELECT t.table_name FROM information_schema.tables AS t "
                                + "WHERE t.table_catalog = $1 AND t.table_schema = $2")
                        .bind("p1")
                        .to(spannerConfig.getDatabaseId().get())
                        .bind("p2")
                        .to("public")
                        .build()))
        .apply(
            MapElements.into(TypeDescriptor.of(ReadOperation.class))
                .via(
                    (SerializableFunction<Struct, ReadOperation>)
                        input -> {
                          String tableName = input.getString(0);
                          return ReadOperation.create()
                              .withQuery("SELECT * FROM \"" + tableName + "\"");
                        }))
        .apply(SpannerIO.readAll().withSpannerConfig(spannerConfig));

Memecahkan masalah kueri yang tidak didukung

Konektor Dataflow hanya mendukung kueri SQL Spanner dengan operator pertama dalam rencana eksekusi kueri adalah Union Terdistribusi. Jika Anda mencoba membaca data dari Spanner menggunakan kueri dan mendapatkan pengecualian yang menyatakan bahwa kueri does not have a DistributedUnion at the root, ikuti langkah-langkah di Memahami cara Spanner menjalankan kueri untuk mengambil rencana eksekusi kueri menggunakan konsol Google Cloud.

Jika kueri SQL Anda tidak didukung, sederhanakan menjadi kueri yang memiliki penggabungan terdistribusi sebagai operator pertama dalam rencana eksekusi kueri. Hapus fungsi agregat, penggabungan tabel, serta operator DISTINCT, GROUP BY, dan ORDER, karena operator tersebut kemungkinan besar akan mencegah kueri berfungsi.

Membuat mutasi untuk operasi tulis

Gunakan metode newInsertOrUpdateBuilder class Mutation, bukan metode newInsertBuilder, kecuali jika benar-benar diperlukan untuk pipeline Java. Untuk pipeline Python, gunakan SpannerInsertOrUpdate, bukan SpannerInsert. Dataflow memberikan jaminan minimal sekali, yang berarti mutasi mungkin ditulis beberapa kali. Akibatnya, mutasi INSERT saja yang dapat menghasilkan error com.google.cloud.spanner.SpannerException: ALREADY_EXISTS yang menyebabkan pipeline gagal. Untuk mencegah error ini, gunakan mutasi INSERT_OR_UPDATE, yang menambahkan baris baru atau memperbarui nilai kolom jika baris sudah ada. Mutasi INSERT_OR_UPDATE dapat diterapkan lebih dari sekali.

Menulis ke Spanner dan mengubah data

Anda dapat menulis data ke Spanner dengan konektor Dataflow menggunakan transformasi SpannerIO.write untuk menjalankan kumpulan mutasi baris input. Konektor Dataflow mengelompokkan mutasi menjadi batch untuk efisiensi.

Contoh berikut menunjukkan cara menerapkan transformasi tulis ke PCollection mutasi:

GoogleSQL 

albums
    // Spanner expects a Mutation object, so create it using the Album's data
    .apply("CreateAlbumMutation", ParDo.of(new DoFn<Album, Mutation>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
        Album album = c.element();
        c.output(Mutation.newInsertOrUpdateBuilder("albums")
            .set("singerId").to(album.singerId)
            .set("albumId").to(album.albumId)
            .set("albumTitle").to(album.albumTitle)
            .build());
      }
    }))
    // Write mutations to Spanner
    .apply("WriteAlbums", SpannerIO.write()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId));

PostgreSQL 

PCollectionView<Dialect> dialectView =
    pipeline.apply(Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton());
albums
    // Spanner expects a Mutation object, so create it using the Album's data
    .apply("CreateAlbumMutation", ParDo.of(new DoFn<Album, Mutation>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
        Album album = c.element();
        c.output(Mutation.newInsertOrUpdateBuilder("albums")
            .set("singerId").to(album.singerId)
            .set("albumId").to(album.albumId)
            .set("albumTitle").to(album.albumTitle)
            .build());
      }
    }))
    // Write mutations to Spanner
    .apply("WriteAlbums", SpannerIO.write()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withDialectView(dialectView));

Jika transformasi tiba-tiba berhenti sebelum selesai, mutasi yang telah diterapkan tidak akan di-roll back.

Menerapkan grup mutasi secara atomik

Anda dapat menggunakan class MutationGroup untuk memastikan bahwa sekelompok mutasi diterapkan secara bersama secara atomik. Mutasi dalam MutationGroup dijamin akan dikirim dalam transaksi yang sama, tetapi transaksi mungkin dicoba lagi.

Grup mutasi berperforma terbaik jika digunakan untuk mengelompokkan mutasi yang memengaruhi data yang disimpan berdekatan di ruang kunci. Karena Spanner menyisipkan data tabel induk dan turunan bersama-sama dalam tabel induk, data tersebut selalu berdekatan dalam ruang kunci. Sebaiknya strukturkan grup mutasi agar berisi satu mutasi yang diterapkan ke tabel induk dan mutasi tambahan yang diterapkan ke tabel turunan, atau agar semua mutasinya mengubah data yang berdekatan di ruang kunci. Untuk informasi selengkapnya tentang cara Spanner menyimpan data tabel induk dan turunan, lihat Skema dan model data. Jika Anda tidak mengatur grup mutasi berdasarkan hierarki tabel yang direkomendasikan, atau jika data yang diakses tidak berdekatan di ruang kunci, Spanner mungkin perlu melakukan commit dua fase, yang menghasilkan performa yang lebih lambat. Untuk informasi selengkapnya, lihat Kompromi lokalitas.

Untuk menggunakan MutationGroup, build transform SpannerIO.write dan panggil metode SpannerIO.Write.grouped, yang menampilkan transformasi yang kemudian dapat Anda terapkan ke PCollection objek MutationGroup.

Saat membuat MutationGroup, mutasi pertama yang tercantum menjadi mutasi utama. Jika grup mutasi Anda memengaruhi tabel induk dan turunan, mutasi utama harus berupa mutasi ke tabel induk. Jika tidak, Anda dapat menggunakan mutasi apa pun sebagai mutasi utama. Konektor Dataflow menggunakan mutasi primer untuk menentukan batas partisi agar dapat mengelompokkan mutasi secara efisien.

Misalnya, bayangkan bahwa aplikasi Anda memantau perilaku dan menandai perilaku pengguna yang bermasalah untuk ditinjau. Untuk setiap perilaku yang ditandai, Anda ingin memperbarui tabel Users untuk memblokir akses pengguna ke aplikasi Anda, dan Anda juga perlu mencatat insiden tersebut di tabel PendingReviews. Untuk memastikan bahwa kedua tabel diperbarui secara atomik, gunakan MutationGroup:

GoogleSQL 

PCollection<MutationGroup> mutations =
    suspiciousUserIds.apply(
        MapElements.via(
            new SimpleFunction<>() {

              @Override
              public MutationGroup apply(String userId) {
                // Immediately block the user.
                Mutation userMutation =
                    Mutation.newUpdateBuilder("Users")
                        .set("id")
                        .to(userId)
                        .set("state")
                        .to("BLOCKED")
                        .build();
                long generatedId =
                    Hashing.sha1()
                        .newHasher()
                        .putString(userId, Charsets.UTF_8)
                        .putLong(timestamp.getSeconds())
                        .putLong(timestamp.getNanos())
                        .hash()
                        .asLong();

                // Add an entry to pending review requests.
                Mutation pendingReview =
                    Mutation.newInsertOrUpdateBuilder("PendingReviews")
                        .set("id")
                        .to(generatedId) // Must be deterministically generated.
                        .set("userId")
                        .to(userId)
                        .set("action")
                        .to("REVIEW ACCOUNT")
                        .set("note")
                        .to("Suspicious activity detected.")
                        .build();

                return MutationGroup.create(userMutation, pendingReview);
              }
            }));

mutations.apply(SpannerIO.write()
    .withInstanceId(instanceId)
    .withDatabaseId(databaseId)
    .grouped());

PostgreSQL 

PCollectionView<Dialect> dialectView =
    pipeline.apply(Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton());
PCollection<MutationGroup> mutations = suspiciousUserIds
    .apply(MapElements.via(new SimpleFunction<String, MutationGroup>() {

      @Override
      public MutationGroup apply(String userId) {
        // Immediately block the user.
        Mutation userMutation = Mutation.newUpdateBuilder("Users")
            .set("id").to(userId)
            .set("state").to("BLOCKED")
            .build();
        long generatedId = Hashing.sha1().newHasher()
            .putString(userId, Charsets.UTF_8)
            .putLong(timestamp.getSeconds())
            .putLong(timestamp.getNanos())
            .hash()
            .asLong();

        // Add an entry to pending review requests.
        Mutation pendingReview = Mutation.newInsertOrUpdateBuilder("PendingReviews")
            .set("id").to(generatedId)  // Must be deterministically generated.
            .set("userId").to(userId)
            .set("action").to("REVIEW ACCOUNT")
            .set("note").to("Suspicious activity detected.")
            .build();

        return MutationGroup.create(userMutation, pendingReview);
      }
    }));

mutations.apply(SpannerIO.write()
    .withInstanceId(instanceId)
    .withDatabaseId(databaseId)
    .withDialectView(dialectView)
    .grouped());

Saat membuat grup mutasi, mutasi pertama yang diberikan sebagai argumen menjadi mutasi utama. Dalam hal ini, kedua tabel tidak terkait, sehingga tidak ada mutasi utama yang jelas. Kita telah memilih userMutation sebagai utama dengan menempatkannya terlebih dahulu. Menerapkan kedua mutasi secara terpisah akan lebih cepat, tetapi tidak akan menjamin atomitas, sehingga grup mutasi adalah pilihan terbaik dalam situasi ini.

Langkah selanjutnya