Mengimpor, mengekspor, dan mengubah data menggunakan Dataflow

Dataflow adalah layanan terkelola untuk mentransformasi dan memperkaya data. Konektor Dataflow untuk Spanner memungkinkan Anda membaca data dari dan menulis data ke Spanner di pipeline Dataflow, yang secara opsional 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 ke dalam dan keluar dari Spanner secara efisien secara massal, dan untuk melakukan transformasi besar ke database yang tidak didukung oleh DML Terpartisi, seperti pemindahan tabel, penghapusan massal yang memerlukan JOIN, dan sebagainya. Ketika menggunakan database individual, ada metode lain yang dapat Anda gunakan untuk mengimpor dan mengekspor data:

  • Gunakan konsol Google Cloud untuk mengekspor database individual dari Spanner ke Cloud Storage dalam format Avro.
  • Gunakan konsol Google Cloud untuk import database kembali ke Spanner dari file yang Anda ekspor ke Cloud Storage.
  • Gunakan REST API atau Google Cloud CLI untuk menjalankan tugas ekspor atau import dari Spanner ke Cloud Storage dan kembali (menggunakan format Avro).

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

Tambahkan 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 Anda 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(). Konfigurasi operasi baca menggunakan metode dalam class SpannerIO.Read. Menerapkan transformasi akan menampilkan PCollection<Struct>, dengan setiap elemen dalam koleksi mewakili satu baris yang ditampilkan oleh operasi baca. Anda dapat membaca dari Spanner dengan dan tanpa kueri SQL tertentu, bergantung pada output yang diinginkan.

Menerapkan transformasi SpannerIO.read() akan menampilkan tampilan data yang konsisten dengan melakukan pembacaan yang kuat. Kecuali jika Anda menentukan sebaliknya, hasil pembacaan akan di-snapshot pada saat Anda memulai pembacaan. Lihat baca 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 dengan 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 sekumpulan kunci utama yang akan dibaca menggunakan metode SpannerIO.Read.withKeySet().

Anda juga dapat membaca tabel menggunakan indeks sekunder yang ditentukan. Seperti pada panggilan API readUsingIndex(), indeks harus berisi semua data yang muncul di hasil kueri.

Untuk melakukannya, tentukan tabel seperti yang ditunjukkan pada contoh sebelumnya, dan tentukan indeks yang berisi nilai kolom yang diinginkan menggunakan metode SpannerIO.Read.withIndex(). Indeks harus menyimpan semua kolom yang perlu dibaca oleh transformasi. {i>Primary key<i} 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 penghentian data transaksi

Transformasi dijamin akan dieksekusi pada snapshot data yang konsisten. Untuk mengontrol status tidak berlaku 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 secara bersamaan untuk memastikan konsistensi data, lakukan semua pembacaan dalam satu transaksi. Untuk melakukannya, terapkan transformasi createTransaction(), dengan 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 Distributed Union. 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 dalam Memahami cara Spanner menjalankan kueri guna mengambil rencana eksekusi untuk kueri Anda menggunakan Konsol Google Cloud.

Jika kueri SQL Anda tidak didukung, sederhanakan kueri yang memiliki gabungan terdistribusi sebagai operator pertama dalam rencana eksekusi kueri. Hapus fungsi agregat, gabungan tabel, serta operator DISTINCT, GROUP BY, dan ORDER, karena mereka adalah operator yang paling mungkin mencegah kueri berjalan.

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 setidaknya satu kali, yang berarti mutasi dapat ditulis beberapa kali. Akibatnya, hanya mutasi INSERT yang mungkin 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 mentransformasi data

Anda dapat menulis data ke Spanner dengan konektor Dataflow menggunakan transformasi SpannerIO.write() untuk menjalankan kumpulan mutasi baris input. Konektor Dataflow mengelompokkan mutasi ke dalam beberapa 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 kelompok mutasi secara atomik

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

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

Untuk menggunakan MutationGroup, build transformasi 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 primer. 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 utama untuk menentukan batas partisi agar dapat mengelompokkan mutasi secara efisien bersama-sama.

Misalnya, bayangkan 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 di tabel PendingReviews. Untuk memastikan kedua tabel diupdate 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 akan menjadi mutasi utama. Dalam hal ini, kedua tabel tersebut tidak terkait, sehingga tidak ada mutasi primer yang jelas. Kita telah memilih userMutation sebagai utama dengan menempatkannya terlebih dahulu. Menerapkan dua mutasi secara terpisah akan lebih cepat, tetapi tidak akan menjamin atomicity, sehingga grup mutasi adalah pilihan terbaik dalam situasi ini.

Langkah selanjutnya