Tüm Ürünler
Search

Bu Ürün

    Realtime Compute for Apache Flink:Veri alımı için Hologres YAML bağlayıcısı

    Belge Merkezi

    Realtime Compute for Apache Flink:Veri alımı için Hologres YAML bağlayıcısı

    Son Güncelleme:Oct 04, 2026

    Bu konu, bir YAML veri alımı dağıtımında verileri eşzamanlamak için Hologres bağlayıcısının nasıl kullanılacağını açıklar.

    Arka plan

    Hologres gerçek zamanlı yazma, güncelleme ve geniş ölçekli analitik işlemlerini destekleyen bir gerçek zamanlı veri ambarı motorudur. Hologres PostgreSQL protokolü ile uyumludur ve standart SQL dilini destekler. Petabayt düzeyindeki veriler üzerinde çevrimiçi analitik işlem (OLAP) ve ad hoc sorguları çalıştırır; ayrıca yüksek eşzamanlılık ve düşük gecikme süresi ile veri sunar. Hologres, MaxCompute, Realtime Compute for Apache Flink ve DataWorks ile entegre çalışarak eksiksiz bir çevrimiçi ve çevrimdışı veri ambarı çözümü sağlar. Aşağıdaki tabloda Hologres YAML bağlayıcısının yetenekleri açıklanmaktadır.

    Öğe

    Açıklama

    Tablo türü

    Sink (Veri hedefi)

    Çalışma modu

    Akış ve toplu iş modları

    Veri biçimi

    Yok

    Ölçüm

    • numRecordsOut

    • numRecordsOutPerSecond

    Not

    Daha fazla bilgi için bkz. Ölçümler.

    API türü

    YAML

    Sink tablosunda güncellemeler veya silmeler

    Desteklenir

    Özellikler


    Özellik


    Açıklama


    Tam veritabanı eşzamanlaması


    Bir veritabanındaki veya birden fazla tablodaki tam ve artımlı verileri gerçek zamanlı olarak ilgili sink tablolarına eşzamanlar.


    Şema değişikliği eşzamanlaması


    Kaynak tablolardaki şema değişikliklerini (sütun ekleme, kaldırma veya yeniden adlandırma gibi) gerçek zamanlı olarak ilgili sink tablolarına yansıtır.


    Bölümlenmiş veritabanı ve tablo eşzamanlaması


    Birden fazla bölümlenmiş veritabanındaki kaynak tabloları adına göre eşleştirmek için normal ifade kullanır. Bu tablolardan gelen veriler birleştirilerek aynı ada sahip downstream sink tablolarına eşzamanlanır.


    Bölümlenmiş tabloya yazma


    Upstream tablodan gelen verileri bir Hologres bölümlenmiş tablosuna yazar.


    Veri türü eşlemesi


    Upstream veri türlerini birden fazla strateji kullanarak daha geniş Hologres veri türlerine eşler.

    Söz dizimi

    sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}

    Parametreler

    Parametre

    Açıklama

    Tür

    Zorunlu

    Varsayılan

    Açıklamalar

    type

    Sink türü.

    String

    Evet

    Yok

    Değer hologres olmalıdır.

    name

    Sink adı.

    String

    Hayır

    Yok

    Yok.

    dbname

    Veritabanı adı.

    String

    Evet

    Yok

    Yok.

    username

    Veritabanına erişim için kullanılan kullanıcı adı. Alibaba Cloud hesabınızın AccessKey ID değerini kullanın.

    String

    Evet

    Yok

    Daha fazla bilgi için bkz. AccessKey ID ve AccessKey Secret değerlerini nasıl görüntülerim?

    Önemli

    AccessKey bilgilerinizin ifşa olmasını önlemek amacıyla değerleri değişkenler aracılığıyla belirtin. Daha fazla bilgi için bkz. Proje değişkenleri.

    password

    Veritabanına erişim için kullanılan parola. Alibaba Cloud hesabınızın AccessKey Secret değerini kullanın.

    String

    Evet

    Yok

    endpoint

    Hologres hizmet uç noktası.

    String

    Evet

    Yok

    Daha fazla bilgi için bkz. Erişim uç noktaları.

    jdbcRetryCount

    Bağlantı başarısız olduğunda yazma ve sorgu işlemleri için yapılacak yeniden deneme sayısı.

    Integer

    Hayır

    10

    Yok.

    jdbcRetrySleepInitMs

    Her yeniden deneme girişimi için sabit bekleme süresi.

    Long

    Hayır

    1000

    Birim: milisaniye. Gerçek yeniden deneme bekleme süresi aşağıdaki formülle hesaplanır: jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs.

    jdbcRetrySleepStepMs

    Her yeniden deneme girişimi için artan bekleme süresi.

    Long

    Hayır

    5000

    Birim: milisaniye. Gerçek yeniden deneme bekleme süresi aşağıdaki formülle hesaplanır: jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs.

    jdbcConnectionMaxIdleMs

    Bir JDBC bağlantısı için maksimum boşta kalma süresi.

    Long

    Hayır

    60000

    Birim: milisaniye. Bir bağlantı bu süreden daha uzun süre boşta kalırsa bağlantı kesilir ve serbest bırakılır.

    jdbcMetaCacheTTL

    Yerel olarak önbelleğe alınan TableSchema bilgilerinin sona erme süresi.

    Long

    Hayır

    60000

    Birim: milisaniye.

    jdbcMetaAutoRefreshFactor

    Önbellek yenileme tetikleyicisini belirleyen çarpan. Kalan ömrü tetikleme süresinden az olan önbellek sistem tarafından otomatik olarak yenilenir.

    Integer

    Hayır

    4

    Kalan önbellek ömrü şu şekilde hesaplanır: Kalan önbellek ömrü = Önbellek sona erme süresi - Önbelleğin aktif olduğu süre. Otomatik bir yenilemeden sonra önbelleğin aktif süresi 0'a sıfırlanır.

    Tetikleme süresi şu formülle hesaplanır: jdbcMetaCacheTTL / jdbcMetaAutoRefreshFactor.

    mutatetype

    Veri yazma modu.

    String

    Hayır

    INSERT_OR_UPDATE

    Hologres fiziksel tablosu için birincil anahtar ayarlanmışsa Hologres sink işlemi birincil anahtara dayalı olarak exactly-once semantiğini garanti eder. Yinelenen birincil anahtara sahip veriler geldiğinde mutatetype parametresi sink tablosunun nasıl güncelleneceğini belirler. mutatetype parametresi aşağıdaki değerleri destekler:

    • INSERT_OR_IGNORE: Verinin ilk oluşumunu korur ve sonraki tüm verileri yok sayar.

    • INSERT_OR_REPLACE: Yeni veri mevcut satırın tamamının yerine geçer.

    • INSERT_OR_UPDATE: Mevcut bir satırın sütunlarının bir alt kümesini günceller. Örneğin, bir tabloda a, b, c ve d olmak üzere dört sütun bulunur ve a Birincil Anahtardır (PK). Yalnızca a ve b sütunları için veriyi Hologres'e yazarsanız ve aynı PK'ya sahip bir satır zaten varsa sistem yalnızca b sütununu günceller. c ve d sütunları değişmez.

    createparttable

    Bölümlenmiş bir tabloya yazma sırasında mevcut olmayan bölümlerin otomatik olarak oluşturulup oluşturulmayacağını belirtir.

    Boolean

    Hayır

    false

    Yok.

    sink.delete-strategy

    Geri çekme mesajlarının nasıl işleneceğini belirtir.

    String

    Hayır

    Yok

    Geçerli değerler:

    • IGNORE_DELETE: Update Before ve Delete mesajlarını yok sayar. Bu değer yalnızca veri eklemeniz veya güncellemeniz gerektiği, silme işlemi yapmanız gerekmediği senaryolar için uygundur.

    • DELETE_ROW_ON_PK: Flink çerçevesi silme işlemini birincil anahtara göre uygular. Güncelleme işlemleri için veri doğruluğunu sağlamak amacıyla önce eski verileri siler ardından yeni verileri ekler.

    jdbcWriteBatchSize

    JDBC modunda toplu yazma işlemi öncesinde Hologres sink tarafında arabelleğe alınacak maksimum kayıt sayısı.

    Integer

    Hayır

    256

    Birim: satır.

    Not

    jdbcWriteBatchSize, jdbcWriteBatchByteSize ve jdbcWriteFlushInterval parametreleri OR (VEYA) ilişkisine sahiptir. Üç parametrenin tamamını ayarlarsanız koşullardan herhangi biri karşılandığında sonuç verisi yazılır.

    jdbcWriteBatchByteSize

    JDBC modunda bu parametre Hologres sink tarafında toplu yazma işlemi öncesinde arabelleğe alınacak verinin bayt cinsinden maksimum boyutunu belirtir.

    Long

    Hayır

    2.097.152 bayt (2 MB)
    Not

    jdbcWriteBatchSize, jdbcWriteBatchByteSize ve jdbcWriteFlushInterval parametreleri OR (VEYA) ilişkisine sahiptir. Üç parametrenin tamamını ayarlarsanız koşullardan herhangi biri karşılandığında sonuç verisi yazılır.

    jdbcWriteFlushInterval

    JDBC modunda bu parametre Hologres sink tarafında arabelleğe alınan verilerin Hologres'e yazılması öncesinde beklenen maksimum süreyi belirtir.

    Long

    Hayır

    10000

    Birim: milisaniye.

    Not

    jdbcWriteBatchSize, jdbcWriteBatchByteSize ve jdbcWriteFlushInterval parametreleri OR (VEYA) ilişkisine sahiptir. Üç parametrenin tamamını ayarlarsanız koşullardan herhangi biri karşılandığında sonuç verisi yazılır.

    ignoreNullWhenUpdate

    mutatetype değeri INSERT_OR_UPDATE olarak ayarlandığında gelen verilerdeki null değerlerin yok sayılıp sayılmayacağını belirtir.

    Boolean

    Hayır

    false

    Geçerli değerler:

    • false (varsayılan): Null değerleri Hologres sink tablosuna yazar.

    • true: Gelen verilerdeki null değerleri yok sayar.

    jdbcEnableDefaultForNotNullColumn

    Tanımlanmış bir varsayılan değeri olmayan NOT NULL bir sütuna null değeri yazıldığında varsayılan bir değerin eklenip eklenmeyeceğini belirtir.

    Boolean

    Hayır

    true

    Geçerli değerler:

    • true (varsayılan): Bağlayıcının varsayılan bir değer eklemesine izin verir. Kurallar şöyledir:

      • String türündeki bir sütun için boş bir dize ("") yazılır.

      • Number türündeki bir sütun için 0 yazılır.

      • Date, timestamp veya timestamptz türündeki sütunlar için 1970-01-01 00:00:00 yazılır.

    • false: Varsayılan bir değer eklemez. NOT NULL bir sütuna null değeri yazıldığında bir özel durum oluşturulur.

    remove-u0000-in-text.enabled

    Yazma işlemi öncesinde dizelerden null karakterinin (\u0000) kaldırılıp kaldırılmayacağını belirtir.

    Boolean

    Hayır

    false

    Geçerli değerler:

    • false (varsayılan): Bağlayıcı veriyi işlemez. Ancak kirli veriyle karşılaşılırsa yazma işlemi sırasında aşağıdaki özel durum oluşabilir: ERROR: invalid byte sequence for encoding "UTF8": 0x00

      Bu durumda kaynak tablodaki kirli verileri işlemeniz veya SQL deyiminizde veri işleme mantığı tanımlamanız gerekir.

    • true: Bağlayıcı yazma özel durumlarını önlemek amacıyla dizelerden \u0000 karakterini kaldırır.

    deduplication.enabled

    jdbc ve jdbc_fixed modlarında yazma işlemi öncesinde her toplu iş içinde yinelemelerin kaldırılıp kaldırılmayacağını belirtir.

    Boolean

    Hayır

    true

    Geçerli değerler:

    • true (varsayılan): Bir toplu iş içinde yinelenen kayıtları etkinleştirir. Birden fazla kayıt aynı birincil anahtara sahipse yalnızca en son kayıt korunur. Örneğin, birinci alanın birincil anahtar olduğu iki alanlı verileri ele alalım:

      • INSERT (1,'a') ve INSERT (1,'b') kayıtları ardışık olarak gelir. Yinelenen kayıtlar kaldırıldıktan sonra Hologres sink tablosuna yalnızca daha sonraki kayıt olan (1,'b') yazılır.

      • Hologres sink tablosunda zaten (1,'a') kaydı bulunmaktadır. DELETE (1,'a') ve INSERT (1,'b') kayıtları ardışık olarak gelirse Hologres'e yalnızca en son gelen kayıt olan (1,'b') yazılır. Bu işlem silme sonrası ekleme değil doğrudan güncelleme olarak kabul edilir.

    • false: Toplu işleme sırasında yinelenen kayıt kaldırma işlemi yapılmaz. Yeni bir kayıt mevcut toplu işteki bir kayıtla aynı birincil anahtara sahipse önce mevcut toplu iş yazılır. Yazma tamamlandıktan sonra yeni kayıt işlenir.

    sink.type-normalize-strategy

    Veri türü eşleme stratejisi.

    String

    Hayır

    STANDARD

    Hologres sink işleminin upstream veri türlerini Hologres türlerine dönüştürmek için kullandığı strateji.

    • STANDARD: Flink CDC türlerini standart PostgreSQL (PG) türlerine dönüştürür.

    • BROADEN: Flink CDC türlerini daha geniş aralıklı Hologres türlerine dönüştürür.

    • ONLY_BIGINT_OR_TEXT: Tüm Flink CDC türlerini Hologres'te BIGINT veya TEXT türüne dönüştürür.

    sink.insert.legacy-put-handler

    Verilerin Hologres'e yazılması için eski Put Handler'ın kullanılıp kullanılmayacağını belirtir.

    Boolean

    Hayır

    false

    Geçerli değerler:

    • false (varsayılan): Veri yazmak için yeni Put Handler kullanılır. Yazma işlemi için SQL biçimi insert into xxx(c0,c1,...) select unnest(?),unnest(?),... on conflict şeklindedir.

    • true: Eski Put Handler kullanılarak veri yazar. Yazma işlemi için SQL biçimi insert into xxx(c0,c1,...) values (?,?,...),... on conflict; şeklindedir.

    table_property.*

    Hologres için fiziksel tablo özellikleri.

    String

    Hayır

    Yok

    Bir Hologres tablosu oluştururken WITH yan tümcesinde fiziksel tablo özelliklerini ayarlayabilirsiniz. Uygun tablo özellikleri sistemin verileri verimli bir şekilde düzenlemesine ve sorgulamasına yardımcı olur.

    Uyarı

    table_property.distribution_key parametresi varsayılan olarak birincil anahtar değerini alır. Veri yazma işleminin doğruluğunu etkileyebileceğinden bu ayarı etkisini tam olarak anlamadığınız sürece değiştirmeyin.

    connection.ssl.mode

    Secure Sockets Layer (SSL) aktarım şifrelemesinin etkinleştirilip etkinleştirilmeyeceğini ve hangi modun kullanılacağını belirtir.

    String

    Hayır

    disable

    • disable (varsayılan): Aktarım şifrelemesini devre dışı bırakır.

    • require: Veri bağlantısını şifrelemek için SSL'i etkinleştirir.

    • verify-ca: SSL'i etkinleştirir, veri bağlantısını şifreler ve Hologres sunucusunun kimliğini doğrulamak için CA sertifikası kullanır.

    • verify-full: SSL'i etkinleştirir, veri bağlantısını şifreler, Hologres sunucusunun kimliğini doğrulamak için CA sertifikası kullanır ve sertifika içindeki CN veya DNS adının yapılandırılan Hologres uç noktasıyla eşleştiğini doğrular.

    Not

    • Hologres V2.1 ve sonraki sürümler verify-ca ve verify-full modlarını destekler. Daha fazla bilgi için bkz. Aktarım şifrelemesi.

    • Bu parametreyi verify-ca veya verify-full olarak ayarlarsanız connection.ssl.root-cert.location parametresini de yapılandırmanız gerekir.

    connection.ssl.root-cert.location

    Aktarım şifreleme modunun sertifika gerektirdiği durumlarda sertifika dosyasının yolunu belirtir.

    String

    Hayır

    Yok

    connection.ssl.mode değeri verify-ca veya verify-full olarak ayarlandığında CA sertifikasının yolunu da yapılandırmanız gerekir. Sertifikayı Realtime Compute konsolundaki Dosya Yönetimi özelliğini kullanarak platforma yükleyebilirsiniz. Sertifika yüklendikten sonra /flink/usrlib dizininde saklanır. Örneğin, CA sertifika dosyasının adı certificate.crt ise parametre değeri '/flink/usrlib/certificate.crt' olmalıdır.

    Not

    CA sertifikasını edinmek için bkz. Aktarım şifrelemesi - CA sertifikası indirme.

    connection.akv4.enabled

    Hologres sunucusuna bağlanmak için AKV4 modunun etkinleştirilip etkinleştirilmeyeceğini belirtir.

    Boolean

    Hayır

    false

    Yok.

    connection.akv4.region

    AKV4 modu etkinleştirildiğinde sunucunun bulunduğu bölgeyi belirtir.

    String

    Hayır

    Yok

    Örneğin cn-shanghai.

    Mevcut bir kataloğu yeniden kullanma

    VVR 11.5 sürümünden itibaren Flink CDC veri alımı işinde Data Management sayfasında oluşturulan yerleşik bir Hologres Catalog'a doğrudan başvurabilirsiniz. Bu yaklaşım bağlantı özelliklerinin manuel olarak belirtilmesi ihtiyacını azaltır.

    sink:
  type: hologres
  using.built-in-catalog: my_holo_catalog

    Veri alımı işleri aşağıdaki Hologres Catalog parametrelerini otomatik olarak yeniden kullanabilir:

    • endpoint

    • username

    • password

    • dbname

    Bu otomatik olarak yeniden kullanılan parametreleri geçersiz kılmak için daha yüksek önceliğe sahip ilgili YAML parametrelerini açıkça belirtebilirsiniz.

    Veri türü eşlemesi

    Upstream verileri Hologres türlerine dönüştürme stratejisini ayarlamak için sink.type-normalize-strategy parametresini kullanabilirsiniz.

    Not

    • Bir YAML işini ilk kez başlatırken sink.type-normalize-strategy seçeneğini etkinleştirin. İş başladıktan sonra etkinleştirirseniz ayarın uygulanması için downstream tabloyu silmeniz ve işi state olmadan yeniden başlatmanız gerekir.

    • Şu anda dizi türleri yalnızca INTEGER, BIGINT, FLOAT, DOUBLE, BOOLEAN, CHAR ve VARCHAR türlerini destekler.

    • Hologres birincil anahtar olarak numeric türünü desteklemez. Birincil anahtarın türü numeric olarak eşlenirse sistem bunu varchar türüne dönüştürür.

    STANDARD

    sink.type-normalize-strategy değeri STANDARD olarak ayarlandığında türler aşağıdaki gibi eşlenir:

    Flink CDC türü

    Hologres türü

    CHAR

    bpchar

    STRING

    text

    VARCHAR

    text (uzunluk > 10485760 ise)

    varchar (uzunluk <= 10485760 ise)

    BOOLEAN

    bool

    BINARY

    bytea

    VARBINARY

    DECIMAL

    numeric

    TINYINT

    int2

    SMALLINT

    INTEGER

    int4

    BIGINT

    int8

    FLOAT

    float4

    DOUBLE

    float8

    DATE

    date

    TIME_WITHOUT_TIME_ZONE

    time

    TIMESTAMP_WITHOUT_TIME_ZONE

    timestamp

    TIMESTAMP_WITH_LOCAL_TIME_ZONE

    timestamptz

    ARRAY

    İlgili öğe türlerinin dizileri

    MAP

    Desteklenmiyor

    ROW

    Desteklenmiyor

    BROADEN

    sink.type-normalize-strategy değeri BROADEN olarak ayarlandığında Flink CDC türleri daha geniş aralıklı Hologres türlerine dönüştürülür. Türler aşağıdaki gibi eşlenir:

    Flink CDC türü

    Hologres türü

    CHAR

    text

    STRING

    VARCHAR

    BOOLEAN

    bool

    BINARY

    bytea

    VARBINARY

    DECIMAL

    numeric

    TINYINT

    int8

    SMALLINT

    INTEGER

    BIGINT

    FLOAT

    float8

    DOUBLE

    DATE

    date

    TIME_WITHOUT_TIME_ZONE

    time

    TIMESTAMP_WITHOUT_TIME_ZONE

    timestamp

    TIMESTAMP_WITH_LOCAL_TIME_ZONE

    timestamptz

    ARRAY

    İlgili öğe türlerinin dizileri

    MAP

    Desteklenmiyor

    ROW

    Desteklenmiyor

    ONLY_BIGINT_OR_TEXT

    sink.type-normalize-strategy değeri ONLY_BIGINT_OR_TEXT olarak ayarlandığında tüm Flink CDC türleri Hologres'te BIGINT veya STRING türlerine dönüştürülür. Veri türü eşlemesi aşağıdaki gibidir:

    Flink CDC türü

    Hologres türü

    TINYINT

    int8

    SMALLINT

    INTEGER

    BIGINT

    BOOLEAN

    text

    BINARY

    VARBINARY

    DECIMAL

    FLOAT

    DOUBLE

    DATE

    TIME_WITHOUT_TIME_ZONE

    TIMESTAMP_WITHOUT_TIME_ZONE

    TIMESTAMP_WITH_LOCAL_TIME_ZONE

    ARRAY

    İlgili öğe türlerinin dizileri

    MAP

    Desteklenmiyor

    ROW

    Desteklenmiyor

    Bölümlenmiş tablolara yazma

    Upstream verileri bir Hologres bölümlenmiş tablosuna yazmak için Hologres sink işlemini transform ile birlikte kullanabilirsiniz.

    • partition key değeri primary key değerinin bir parçası olmalıdır. Upstream verilerden birincil anahtar olmayan bir sütunu partition key olarak kullanırsanız upstream ve downstream tabloların birincil anahtarları tutarsız hale gelebilir ve bu durum veri eşzamanlaması sırasında veri farklılıklarına neden olabilir.

    • Hologres TEXT, VARCHAR ve INT veri türlerindeki sütunların partition key olarak kullanılmasını destekler. 1.3.22 sürümünden itibaren DATE veri türündeki sütunlar da desteklenmektedir.

    • Alt bölümlenmiş tabloları otomatik olarak oluşturmak için createparttable parametresini true olarak ayarlayın. Aksi takdirde bunları manuel olarak oluşturmanız gerekir.

    Örnek için bkz. Bölümlenmiş tabloya veri yazma.

    Tablo şeması eşzamanlaması

    CDC YAML işlem hattı tablo şeması değişikliklerini işlemek için farklı stratejiler kullanır ve bu stratejiler işlem hattı düzeyindeki schema.change.behavior parametresiyle yapılandırılabilir. schema.change.behavior için geçerli değerler IGNORE, LENIENT, TRY_EVOLVE, EVOLVE ve EXCEPTION değerleridir. Hologres Sink şu anda TRY_EVOLVE stratejisini desteklememektedir. LENIENT ve EVOLVE stratejileri tablo şeması değişikliklerini içerir. Aşağıdaki bölümlerde bu iki modun farklı şema değişikliği olaylarını nasıl işlediği açıklanmaktadır.

    LENIENT (varsayılan)

    LENIENT modunda şema değişiklikleri aşağıdaki gibi işlenir:

    • Boş bırakılabilen bir sütun ekleme: İlgili sütun sink tablosunun sonuna otomatik olarak eklenir ve verisi eşzamanlanır.

    • Boş bırakılabilen bir sütun silme: Sütun sink tablosundan silinmez. Bunun yerine sütun otomatik olarak NULL değerlerle doldurulur.

    • Boş bırakılamayan bir sütun ekleme: İlgili boş bırakılabilir sütun sink tablosunun sonuna otomatik olarak eklenir ve verisi eşzamanlanır. Mevcut satırlar için bu yeni sütun otomatik olarak NULL ile doldurulur.

    • Bir sütunu yeniden adlandırma: Bu işlem bir sütunu silme ve yeni bir sütun ekleme olarak değerlendirilir. Belirtilen ada sahip yeni bir sütun sink tablosunun sonuna eklenir ve orijinal ada sahip sütun otomatik olarak NULL değerlerle doldurulur. Örneğin col_a sütunu col_b olarak yeniden adlandırılırsa sink tablosunun sonuna bir col_b sütunu eklenir ve col_a sütunu otomatik olarak NULL değerlerle doldurulur.

    • Bir sütunun veri türünü değiştirme: Desteklenmez. Hologres bir sütunun veri türünün değiştirilmesini desteklemediğinden sink.type-normalize-strategy parametresini kullanmanız gerekir.

    • Aşağıdaki şema değişiklikleri desteklenmez:

      • Birincil anahtar veya dizin gibi kısıtlamalardaki değişiklikler.

      • Boş bırakılamayan bir sütunu silme.

      • Bir sütunu NOT NULL'dan NULLABLE'a değiştirme.

    EVOLVE

    EVOLVE modunda şema değişiklikleri aşağıdaki gibi işlenir:

    • Boş bırakılabilen bir sütun ekleme: Desteklenir.

    • Boş bırakılabilen bir sütun silme: Desteklenmez.

    • Boş bırakılamayan bir sütun ekleme: Sink tablosuna yeni bir boş bırakılabilir sütun eklenir.

    • Bir sütunu yeniden adlandırma: Desteklenir. Orijinal sütun sink tablosunda yeniden adlandırılır.

    • Bir sütunun veri türünü değiştirme: Desteklenmez. Hologres bir sütunun veri türünün değiştirilmesini desteklemediğinden sink.type-normalize-strategy parametresini kullanmanız gerekir.

    • Aşağıdaki şema değişiklikleri desteklenmez:

      • Birincil anahtar veya dizin gibi kısıtlamalardaki değişiklikler.

      • Boş bırakılamayan bir sütunu silme.

      • Bir sütunu NOT NULL'dan NULLABLE'a değiştirme.

    Uyarı

    EVOLVE modunda sink tablosunu silmeden stateless yeniden başlatma işlemi yapmak upstream veriler ile sink tablosu arasındaki şema tutarsızlıkları nedeniyle işlem hattının başarısız olmasına neden olabilir. Bu durumda sink tablosu şemasını manuel olarak ayarlamanız gerekir.

    Örnek için bkz. EVOLVE modunu etkinleştirme.

    Kod örnekleri

    Tür genişletme

    Tür genişletmeyi yapılandırmak için sink.type-normalize-strategy parametresini kullanın.

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    Bölümlenmiş tabloya yazma

    create_time timestamp alanını date türüne dönüştürün ve bunu Hologres tablosu için partition key olarak kullanın.

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Hologres Pipeline

    EVOLVE modunu etkinleştirme

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true

pipeline:
  name: MySQL to Hologres Pipeline
  schema.change.behavior: evolve

    Tek tablo eşzamanlaması

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    Tam veritabanı eşzamanlaması

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    Bölümlenmiş tabloları birleştirme

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.  
  sink.type-normalize-strategy: BROADEN
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Hologres table named test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

    Belirtilen bir şemaya eşzamanlama

    Hologres'te bir şema MySQL'deki bir veritabanına karşılık gelir. Sink tabloları için şemayı belirtebilirsiniz.

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN
  
route:
  # Synchronize all tables from the MySQL test_db database to the Hologres test_db2 schema while keeping the original table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

    Yeniden başlatma olmadan yeni tabloları eşzamanlama

    Bir iş çalışırken yeni eklenen tabloları gerçek zamanlı olarak eşzamanlamak için scan.binlog.newly-added-table.enable = true ayarını yapın.

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # Automatically captures new tables created while the job is running.  
  scan.binlog.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    Yeniden başlatmada mevcut tabloları ekleme

    Mevcut bir tabloyu eşzamanlamaya dahil etmek için scan.newly-added-table.enabled değerini true olarak ayarlayın ve işi yeniden başlatın.

    Uyarı

    Daha önce scan.binlog.newly-added-table.enabled ile çalıştırılmış bir işte scan.newly-added-table.enabled kullanmayın. Bu kombinasyon yeniden başlatma sırasında veri yinelenmesine neden olur.

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On restart, the job scans for new tables matching the `tables` parameter and performs a snapshot.
  # Note: This parameter must be used with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    Tabloları hariç tutma

    source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Excludes tables that match this regular expression.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

    İlgili belgeler