Pengembangan dan debugging - Realtime Compute for Apache Flink

Topik ini mencakup isu umum dalam pengembangan dan debugging.

Bagaimana cara mendeklarasikan pernyataan DDL saat mengirimkannya bersama pernyataan DML dalam skrip yang sama?
Bagaimana cara menulis beberapa pernyataan INSERT INTO?
Bagaimana cara meneruskan karakter khusus dalam Entry point Main Arguments?
Mengapa unggahan paket JAR UDF gagal setelah beberapa kali modifikasi?
Mengapa bidang tidak sejajar saat kelas POJO digunakan sebagai tipe kembalian UDTF?
Bagaimana cara mengatasi konflik dependensi Flink?
Error: Could not parse type at position 50: expected but was . Input type string: ROW
Error saat menulis ke tabel: "Invalid primary key. Column 'xxx' is nullable."
Mengapa file JSON terbuka di browser alih-alih diunduh?

Deklarasi DDL bersama pernyataan DML

Saat Anda mengirimkan pernyataan DDL dan DML bersama dalam skrip yang sama, deklarasikan pernyataan DDL sebagai CREATE TEMPORARY TABLE alih-alih CREATE TABLE. Jika tidak, klik tombol Validate akan gagal dengan error seperti berikut.

Pesan kesalahan:CREATE TABLE datagen_source (a bigint, b int, c varchar)
  WITH ('connector' = 'datagen');
CREATE  TABLE print_sink(C bigint, var1 int)
  WITH ('connector' = 'print','logger' = 'true');
INSERT INTO print_sink SELECT a,8  FROM datagen_source;
<code code-type="xCode" data-tag="codeblock">Error message:

Beberapa pernyataan INSERT INTO

Untuk membentuk satu unit logis, bungkus beberapa pernyataan INSERT INTO di antara BEGIN STATEMENT SET; dan END;. Untuk informasi selengkapnya, lihat INSERT INTO statements. Jika Anda tidak membungkus pernyataan tersebut, klik tombol Validate akan gagal dengan error seperti berikut.

CREATE TEMPORARY TABLE datagen_source (a bigint, b int, c varchar)
  WITH ('connector' = 'datagen');
CREATE TEMPORARY TABLE print_sink (C bigint, var1 int)
  WITH ('connector' = 'print', 'logger' = 'true');
CREATE TEMPORARY TABLE print_sink2 (C bigint, var2 int)
  WITH ('connector' = 'print', 'logger' = 'true');
INSERT INTO print_sink SELECT a, B FROM datagen_source;
INSERT INTO print_sink2 SELECT a, B FROM datagen_source;
org.apache.flink.table.gateway.api.vvr.utils.SqlValidationException: A sequence of multiple statements to execute is supported if the last statement is a 'SELECT' statement or 'INSERT INTO' statement or 'CREATE TABLE IF NOT EXISTS ... AS TABLE' statement or 'CREATE DATABASE IF NOT EXISTS ... AS DATABASE' statement or 'AUTO OPTIMIZE TABLE|DATABASE' statements or multiple 'INSERT INTO' or 'CREATE TABLE IF NOT EXISTS ... AS TABLE' or 'CREATE DATABASE IF NOT EXISTS ... AS DATABASE' statements wrapped in a 'BEGIN STATEMENT SET' block and all other statements are CREATE TEMPORARY TABLE|VIEW|[SYSTEM] FUNCTION, 'SHOW', DESCRIBE, 'USE' statements.
      at org.apache.flink.table.gateway.vvr.service.utils.SqlValidateUtils.validateDraft(SqlValidateUtils.java:79)
      at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.getDraftType(DraftCommand.java:120)
      at org.apache.flink.table.gateway.vvr.service.command.DraftCommand.executeInternal(DraftCommand.java:71)
      at java.security.AccessController.doPrivileged(Native Method)
      at javax.security.auth.Subject.doAs(Subject.java:422)
      at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1899)
      at org.apache.flink.table.gateway.service.context.SqlGatewaySecurityContext.runSecured(SqlGatewaySecurityContext.java:73)
      at org.apache.flink.table.gateway.vvr.service.command.AbstractCommand.wrapClassLoader(AbstractCommand.java:171)
      at org.apache.flink.table.gateway.vvr.service.command.AbstractCommand.execute(AbstractCommand.java:163)
      at org.apache.flink.table.gateway.vvr.service.command.CommandManager.lambda$execute$0(CommandManager.java:71)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
      at java.util.concurrent.FutureTask.run(FutureTask.java:266)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

Meneruskan karakter khusus dalam argumen main

Penyebab

Saat Anda meneruskan karakter khusus seperti # dan $ dalam Entry point Main Arguments, karakter escape backslash (\) tidak berfungsi dan karakter khusus tersebut dibuang.
Solusi

Pada halaman Deployments, klik nama deployment target. Di bagian Parameters, tambahkan parameter env.java.opts: -Dconfig.disable-inline-comment=true ke kolom Other Configuration. Untuk informasi selengkapnya, lihat How to configure custom deployment parameters.

Gagal unggah JAR UDF setelah modifikasi

Penyebab

Runtime UDF mewajibkan nama kelas unik di seluruh paket JAR.
Solusi
- Hapus paket JAR lama dan unggah yang baru.
- Unggah paket JAR di bagian Additional Dependencies dan gunakan fungsi temporary dalam kode Anda. Untuk informasi tentang cara menggunakan fungsi temporary, lihat Register a UDF. Contoh berikut menunjukkan sintaksnya.
```
CREATE TEMPORARY FUNCTION `cp_record_reduce` AS 'com.taobao.test.udf.blink.CPRecordReduceUDF';
```
  Di bagian Additional Dependencies, yang terletak di panel Other Configuration sebelah kanan, masukkan URL OSS dari paket JAR UDF Anda.

Ketidaksejajaran bidang dengan kelas POJO sebagai tipe kembalian UDTF

Detail

Saat Anda menggunakan kelas POJO sebagai tipe kembalian untuk UDTF dan secara eksplisit mendeklarasikan daftar alias untuk kolom yang dikembalikan dalam SQL, Anda mungkin mengalami masalah ketidaksejajaran bidang. Akibatnya, bidang aktual mungkin bukan yang Anda harapkan, meskipun tipe datanya konsisten.

Sebagai contoh, jika Anda menggunakan kelas POJO berikut sebagai tipe kembalian untuk UDTF, memaketkannya, dan mendaftarkan fungsi tersebut sebagai deployment-level UDF seperti yang dijelaskan dalam Develop a custom function, validasi SQL akan gagal.

package com.aliyun.example;
public class TestPojoWithoutConstructor {
	public int c;
	public String d;
	public boolean a;
	public String b;
}

package com.aliyun.example;
import org.apache.flink.table.functions.TableFunction;
public class MyTableFuncPojoWithoutConstructor extends TableFunction<TestPojoWithoutConstructor> {
	private static final long serialVersionUID = 1L;
	public void eval(String str1, Integer i2) {
		TestPojoWithoutConstructor p = new TestPojoWithoutConstructor();
		p.d = str1 + "_d";
		p.c = i2 + 2;
		p.b = str1 + "_b";
		collect(p);
	}
}

CREATE TEMPORARY FUNCTION MyTableFuncPojoWithoutConstructor as 'com.aliyun.example.MyTableFuncPojoWithoutConstructor';
CREATE TEMPORARY TABLE src ( 
  id STRING,
  cnt INT
) WITH (
  'connector' = 'datagen'
);
CREATE TEMPORARY TABLE sink ( 
  f1 INT,
  f2 STRING,
  f3 BOOLEAN,
  f4 STRING
) WITH (
 'connector' = 'print'
);
INSERT INTO sink
SELECT T.* FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b);

Validasi SQL mengembalikan pesan error berikut:

org.apache.flink.table.api.ValidationException: SQL validation failed. Column types of query result and sink for 'vvp.default.sink' do not match.  
Cause: Sink column 'f1' at position 0 is of type INT but expression in the query is of type BOOLEAN NOT NULL.  
Hint: You will need to rewrite or cast the expression.  
Query schema: [c: BOOLEAN NOT NULL, d: STRING, a: INT NOT NULL, b: STRING]  
Sink schema:  [f1: INT, f2: STRING, f3: BOOLEAN, f4: STRING]  
	at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)

Bidang yang dikembalikan dari UDTF tidak sejajar dengan bidang dalam kelas POJO. Dalam hasil kueri, bidang c bertipe BOOLEAN dan bidang a bertipe INT, yang merupakan kebalikan dari definisi mereka dalam kelas POJO.

Penyebab

Berdasarkan aturan inferensi tipe untuk kelas POJO:
- Jika kelas POJO memiliki konstruktor berparameter, Flink melakukan inferensi tipe kembalian berdasarkan urutan parameter konstruktor.
- Jika kelas POJO tidak memiliki konstruktor berparameter, Flink mengurutkan ulang bidang berdasarkan urutan alfabetis nama bidang.
Dalam contoh ini, karena kelas POJO yang digunakan sebagai tipe kembalian UDTF tidak memiliki konstruktor berparameter, bidang dikembalikan dalam urutan alfabetis, menghasilkan tipe BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d). Meskipun inferensi ini valid, kueri SQL secara eksplisit mengganti nama kolom output dengan LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b). Daftar alias ini mengganti nama kolom berdasarkan posisi, sehingga terjadi ketidaksesuaian dengan bidang yang telah diurutkan secara alfabetis dari POJO. Konflik antara alias berbasis posisi dan pengurutan bidang alfabetis ini menyebabkan exception validasi atau ketidaksejajaran data yang tidak diinginkan.

Solusi

Jika kelas POJO tidak memiliki konstruktor berparameter, hapus penggantian nama eksplisit pada bidang kembalian UDTF. Sebagai contoh, ubah pernyataan INSERT dalam SQL menjadi:

-- Jika kelas POJO tidak memiliki konstruktor berparameter, pilih bidang yang diperlukan berdasarkan nama.  
-- Saat menggunakan T.*, perhatikan urutan aktual bidang yang dikembalikan.  
SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T;

Implementasikan konstruktor berparameter dalam kelas POJO untuk mengontrol urutan bidang dalam tipe kembalian. Dalam kasus ini, urutan bidang output UDTF akan sesuai dengan urutan parameter konstruktor.

package com.aliyun.example;
public class TestPojoWithConstructor {
	public int c;
	public String d;
	public boolean a;
	public String b;
	// Menggunakan urutan bidang tertentu alih-alih urutan alfabetis
	public TestPojoWithConstructor(int c, String d, boolean a, String b) {
		this.c = c;
		this.d = d;
		this.a = a;
		this.b = b;
	}
}

Mengatasi konflik dependensi Flink

Gejala
- Konflik muncul sebagai error jelas yang dilemparkan oleh kelas terkait Flink atau Hadoop.
```
java.lang.AbstractMethodError
java.lang.ClassNotFoundException
java.lang.IllegalAccessError
java.lang.IllegalAccessException
java.lang.InstantiationError
java.lang.InstantiationException
java.lang.InvocationTargetException
java.lang.NoClassDefFoundError
java.lang.NoSuchFieldError
java.lang.NoSuchFieldException
java.lang.NoSuchMethodError
java.lang.NoSuchMethodException
```
- Atau, masalah dapat muncul sebagai perilaku tak terduga tanpa pesan error yang jelas, seperti:
  - Log tidak dihasilkan atau konfigurasi log4j tidak berlaku.
    
    Masalah ini biasanya disebabkan oleh konfigurasi terkait log4j yang termasuk dalam dependensi. Periksa apakah paket JAR deployment berisi dependensi yang membawa konfigurasi log4j. Anda dapat menghapus konfigurasi ini dengan menggunakan exclusions dalam definisi dependensi Anda.
    
    Catatan
    Jika Anda harus menggunakan versi log4j yang berbeda, gunakan maven-shade-plugin untuk mereposisi kelas-kelas terkait log4j.
  - Exception panggilan RPC.
    
    Konflik dependensi yang memengaruhi panggilan RPC Akka Flink dapat menyebabkan exception yang tidak ditampilkan di log secara default. Anda harus mengaktifkan logging debug untuk mengidentifikasinya.
    
    Sebagai contoh, log debug menunjukkan Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}, tetapi log JobManager (JM) tidak menunjukkan aktivitas apa pun setelah Registering TaskManager with ResourceID xxx hingga terjadi error timeout NoResourceAvailableException. Sementara itu, TaskManager (TM) terus-menerus melaporkan error Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}.
    
    Penyebab: Dengan logging debug diaktifkan, Anda dapat melihat bahwa InvocationTargetException dilemparkan selama panggilan RPC. Error ini menyebabkan alokasi slot TM gagal di tengah proses, menghasilkan keadaan tidak konsisten. ResourceManager (RM) kemudian terus-menerus dan gagal mencoba mengalokasikan slot, serta tidak dapat pulih.
Penyebab
- Paket JAR deployment berisi dependensi yang tidak perlu, seperti library dasar Flink, Hadoop, atau log4j, yang menyebabkan konflik dependensi.
- Dependensi untuk konektor yang diperlukan tidak disertakan dalam paket JAR.
Pemecahan Masalah
- Tinjau file pom.xml deployment untuk dependensi yang tidak perlu.
- Periksa isi paket JAR deployment dengan menjalankan perintah jar tf foo.jar untuk memeriksa file yang bertentangan.
- Analisis pohon dependensi deployment untuk konflik dengan menjalankan perintah mvn dependency:tree.

Solusi

Sebagai praktik terbaik, atur scope dependensi framework dasar ke provided. Hal ini mencegah dependensi tersebut dimasukkan ke dalam paket JAR deployment.

DataStream Java

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>

DataStream Scala

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-scala_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>

DataSet Java

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-java</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>

DataSet Scala

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-scala_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>

Tambahkan dependensi konektor yang diperlukan ke proyek Anda. Scope default adalah compile, yang secara benar memasukkannya ke dalam paket JAR deployment. Sebagai contoh, untuk menambahkan konektor Kafka:
```
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
```
Jangan menambahkan dependensi Flink, Hadoop, atau log4j lainnya. Namun:
- Jika deployment memiliki dependensi langsung pada komponen konfigurasi dasar atau terkait konektor, atur scope-nya ke provided. Contoh berikut menunjukkan sintaksnya.
```
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <scope>provided</scope>
</dependency>
```
- Jika deployment memiliki dependensi transitif pada komponen konfigurasi dasar atau terkait konektor, hapus dengan menggunakan exclusion. Contoh berikut menunjukkan sintaksnya.
```
<dependency>
    <groupId>foo</groupId>
    <artifactId>bar</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
        </exclusion>
    </exclusions>
</dependency>
```

Error: Could not parse type at position 50: expected but was . Input type string: ROW

Detail kesalahan

Saat menulis SQL di editor SQL, terjadi error pemeriksaan sintaks (garis bergelombang merah) saat Anda menggunakan UDTF.

Caused by: org.apache.flink.table.api.ValidationException: Could not parse type at position 50: <IDENTIFIER> expected but was <KEYWORD>. Input type string: ROW<resultId String,pointRange String,from String,to String,type String,pointScope String,userId String,point String,triggerSource String,time String,uuid String>

Kode berikut adalah contohnya:

@FunctionHint(
    //input = @DataTypeHint("BYTES"),
    output = @DataTypeHint("ROW<resultId String,pointRange String,from String,to String,type String,pointScope String,userId String,point String,triggerSource String,time String,uuid String>"))
public class PointChangeMetaQPaser1 extends TableFunction<Row> {
    Logger logger = LoggerFactory.getLogger(this.getClass().getName());
    public void eval(byte[] bytes) {
        try {
            String messageBody = new String(bytes, "UTF-8");
            Map<String, String> resultDO = JSON.parseObject(messageBody, Map.class);
            logger.info("PointChangeMetaQPaser1 logger:" + JSON.toJSONString(resultDO));
            collect(Row.of(
                    getString(resultDO.get("resultId")),
                    getString(resultDO.get("pointRange")),
                    getString(resultDO.get("from")),
                    getString(resultDO.get("to")),
                    getString(resultDO.get("type")),
                    getString(resultDO.get("pointScope")),
                    getString(resultDO.get("userId")),
                    getString(resultDO.get("point")),
                    getString(resultDO.getOrDefault("triggerSource", "NULL")),
                    getString(resultDO.getOrDefault("time", String.valueOf(System.currentTimeMillis()))),
                    getString(resultDO.getOrDefault("uuid", String.valueOf(UUID.randomUUID())))
            ));
        } catch (Exception e) {
            logger.error("PointChangeMetaQPaser1 error", e);
        }
    }
    private String getString(Object o) {
        if (o == null) {
            return null;
        }
        return String.valueOf(o);
    }
}

Penyebab

Saat Anda menggunakan DataTypeHint untuk mendefinisikan tipe data fungsi, kata kunci yang dicadangkan digunakan langsung sebagai nama bidang.
Solusi
- Ubah nama bidang menjadi non-keyword. Sebagai contoh, ubah to menjadi fto dan from menjadi ffrom.
- Bungkus bidang yang menggunakan kata kunci yang dicadangkan dengan backtick ().

Error: "Invalid primary key. Column 'xxx' is nullable."

Penyebab

Flink mewajibkan semua kolom kunci primer dideklarasikan secara eksplisit sebagai NOT NULL. Bahkan jika data tidak mengandung nilai NULL, Flink menolak operasi tersebut sebelum penulisan jika kolom kunci primer dalam pernyataan pembuatan tabel mengizinkan nilai NULL (misalnya, INT NULL). Ini bukan error runtime, melainkan pemeriksaan semantik selama fase parsing DDL.
Solusi

Deklarasikan kolom kunci primer yang disebutkan dalam error sebagai NOT NULL dan buat ulang tabel tersebut.

File JSON terbuka di browser alih-alih diunduh

Gejala

Saat Anda mengklik untuk mengunduh file JSON dari halaman Artifacts, browser tidak memicu unduhan. Sebaliknya, browser membuka tab baru dan menampilkan konten JSON secara langsung.
Penyebab

File JSON di OSS tidak memiliki header respons HTTP Content-Disposition: attachment. Hal ini menyebabkan browser menampilkan konten file secara langsung alih-alih mengunduhnya.
Solusi
- Opsi 1: Unggah ulang file
  
  Masalah ini telah diperbaiki di versi platform 4.5.0, tetapi perbaikan hanya berlaku untuk file yang diunggah setelah rilis versi ini. File yang diunggah sebelum tanggal tersebut harus ditangani secara manual.
- Opsi 2: Modifikasi metadata objek OSS
  
  Modifikasi metadata objek secara manual dengan menambahkan atribut HTTP standar berikut:
  - Nama header: Content-Disposition
  - Nilai header: attachment
  Untuk informasi selengkapnya, lihat Manage object metadata.