このトピックでは、開発とデバッグに関する一般的な問題について説明します。
DML ステートメントと DDL の宣言
DDL ステートメントと DML ステートメントを同じスクリプトで送信する場合、DDL ステートメントは CREATE TABLE ではなく CREATE TEMPORARY TABLE として宣言します。そうしないと、[Validate] をクリックしたときに次のようなエラーが発生して失敗します。
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;
エラーメッセージ：
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 ...BASE IF NOT EXISTS ... AS DATABASE' statement or 'AUTO OPTIMIZE TABLE|DATABASE' statements or multiple 'INSERT INTO' or 'CREATE TABLE IF NOT EXISTS ... AS TAB ...ATABASE 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:107)
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)複数の INSERT INTO ステートメント
複数の INSERT INTO ステートメントを 1 つの論理単位として構成するには、ステートメントを BEGIN STATEMENT SET; と END; で囲みます。詳細については、「INSERT INTO ステートメント」をご参照ください。ステートメントを囲まない場合、[Validate] をクリックしたときに次のようなエラーが発生して失敗します。
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)Main 引数での特殊文字の受け渡し
-
原因
エントリーポイントの Main 引数で # や $ などの特殊文字を渡す場合、バックスラッシュ (\) でエスケープできず、特殊文字は破棄されます。
-
解決策
[Deployments] ページで、対象のデプロイメント名をクリックします。 [Parameters] セクションで、[Other Configuration] フィールドにパラメータ
env.java.opts: -Dconfig.disable-inline-comment=trueを追加します。詳細については、「カスタムデプロイメントパラメータの構成方法」をご参照ください。
変更後の UDF JAR アップロード失敗
-
原因
UDF ランタイムは、JAR パッケージ間で一意のクラス名を強制します。
-
解決策
-
古い JAR パッケージを削除してから、新しいパッケージをアップロードします。
-
[Additional Dependencies] セクションで JAR パッケージをアップロードし、コード内で一時関数を使用します。一時関数の使用方法については、「UDF の登録」をご参照ください。以下に構文の例を示します。
CREATE TEMPORARY FUNCTION `cp_record_reduce` AS 'com.taobao.test.udf.blink.CPRecordReduceUDF';右側の [Other Configuration] パネルにある [Additional Dependencies] セクションで、UDF JAR パッケージの OSS URL を指定します。
-
POJO クラスを UDTF の戻り値の型として使用した場合のフィールドのずれ
-
詳細
POJO クラスを UDTF の戻り値の型として使用し、SQL で返されるカラムのエイリアスリストを明示的に宣言すると、フィールドのずれの問題が発生する可能性があります。その結果、データ型に一貫性があっても、実際のフィールドが期待するものと異なる場合があります。
たとえば、次の POJO クラスを UDTF の戻り値の型として使用し、パッケージ化して、「カスタム関数の開発」で説明されているように関数を デプロイメントレベルの UDF として登録すると、SQL の検証が失敗します。
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);SQL の検証では、次のエラーメッセージが返されます。
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)UDTF から返されるフィールドは、POJO クラスのフィールドとずれています。クエリ結果では、フィールド
cは BOOLEAN 型で、フィールドaは INT 型です。これは、POJO クラスでの定義とは逆になっています。 -
原因
POJO クラスの型推論ルールによると、次のようになります。
-
POJO クラスにパラメーター化されたコンストラクターがある場合、Flink はコンストラクターのパラメータの順序に基づいて戻り値の型を推論します。
-
POJO クラスにパラメーター化されたコンストラクターがない場合、Flink はフィールド名のアルファベット順にフィールドを並べ替えます。
この例では、UDTF の戻り値の型に使用される POJO クラスにパラメーター化されたコンストラクターがないため、フィールドはアルファベット順に返され、型は
BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d)になります。この推論は有効ですが、SQL クエリはLATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b)で出力カラムを明示的に名前変更しています。このエイリアスリストは位置によってカラムの名前を変更するため、POJO のアルファベット順のフィールドとの不一致が生じます。位置によるエイリアシングとアルファベット順のフィールド順序との間のこの競合により、検証例外または予期しないデータのずれが発生します。 -
-
解決策
-
POJO クラスにパラメーター化されたコンストラクターがない場合は、UDTF の戻りフィールドの明示的な名前変更を削除します。たとえば、SQL の INSERT ステートメントを次のように変更します。
-- POJO クラスにパラメーター化されたコンストラクターがない場合は、必要なフィールドを名前で選択します。 -- T.* を使用する場合は、返されるフィールドの実際の順序を認識している必要があります。 SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T; -
POJO クラスにパラメーター化されたコンストラクターを実装して、戻り値の型のフィールドの順序を制御します。この場合、UDTF の出力のフィールド順序は、コンストラクターのパラメータの順序と一致します。
package com.aliyun.example; public class TestPojoWithConstructor { public int c; public String d; public boolean a; public String b; // アルファベット順ではなく、特定のフィールド順序を使用 public TestPojoWithConstructor(int c, String d, boolean a, String b) { this.c = c; this.d = d; this.a = a; this.b = b; } }
-
Flink の依存関係の競合の解決
-
症状
-
競合は、Flink または 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 -
または、明確なエラーメッセージがなく、次のような予期しない動作として問題が現れる場合があります。
-
ログが生成されないか、log4j の構成が有効になりません。
この問題は通常、依存関係に含まれる log4j 関連の構成が原因で発生します。デプロイメント JAR パッケージに log4j 構成を含む依存関係が含まれているかどうかを確認してください。依存関係定義で除外を使用して、これらの構成を削除できます。
説明異なるバージョンの log4j を使用する必要がある場合は、maven-shade-plugin を使用して log4j 関連のクラスを再配置してください。
-
RPC 呼び出しの例外。
Flink の Akka RPC 呼び出しに影響する依存関係の競合により、デフォルトではログに表示されない例外が発生する可能性があります。これらを識別するには、デバッグログを有効にする必要があります。
たとえば、デバッグログには
Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}と表示されますが、JobManager (JM) ログにはRegistering TaskManager with ResourceID xxxの後にアクティビティが表示されず、NoResourceAvailableExceptionタイムアウトエラーが発生するまで続きます。一方、TaskManager (TM) は継続的にCannot allocate the requested resources. Trying to allocate ResourceProfile{xxx}というエラーを報告します。原因: デバッグログを有効にすると、RPC 呼び出し中に
InvocationTargetExceptionがスローされていることがわかります。このエラーにより、TM スロット割り当てがプロセスの途中で失敗し、不整合な状態になります。その後、ResourceManager (RM) は継続的にスロットの割り当てを試みますが失敗し、回復できなくなります。
-
-
-
原因
-
デプロイメント JAR パッケージに、基本的な Flink、Hadoop、または log4j ライブラリなどの不要な依存関係が含まれており、依存関係の競合が発生しています。
-
必要なコネクタの依存関係が JAR パッケージに含まれていません。
-
-
トラブルシューティング
-
デプロイメントの
pom.xmlファイルで不要な依存関係を確認します。 -
jar tf foo.jarコマンドを実行して、デプロイメント JAR パッケージの内容を調べ、競合するファイルを確認します。 -
mvn dependency:treeコマンドを実行して、デプロイメントの依存関係ツリーを分析し、競合を確認します。
-
-
解決策
-
ベストプラクティスとして、基本的なフレームワークの依存関係の
スコープをprovidedに設定します。これにより、デプロイメント JAR パッケージにバンドルされなくなります。-
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>
-
-
必要なコネクタの依存関係をプロジェクトに追加します。デフォルトのスコープは
compileで、デプロイメント JAR パッケージに正しくバンドルされます。たとえば、Kafka コネクタを追加するには、次のようにします。<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>${flink.version}</version> </dependency> -
その他の Flink、Hadoop、または log4j の依存関係は追加しないでください。ただし、次の場合は例外です。
-
デプロイメントが基本構成またはコネクタ関連コンポーネントに直接依存している場合は、スコープを
providedに設定します。以下に構文の例を示します。<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <scope>provided</scope> </dependency> -
デプロイメントが基本構成またはコネクタ関連コンポーネントに推移的な依存関係を持っている場合は、
exclusionを使用して削除します。以下に構文の例を示します。<dependency> <groupId>foo</groupId> <artifactId>bar</artifactId> <exclusions> <exclusion> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> </exclusion> </exclusions> </dependency>
-
-
エラー: Could not parse type at position 50: expected but was . Input type string: ROW
-
エラーの詳細
SQL エディタで SQL を記述する際、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>以下はコードの例です。
@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); } } -
原因
DataTypeHint を使用して関数のデータ型を定義する際、予約キーワードがフィールド名として直接使用されています。
-
解決策
-
フィールド名をキーワード以外に変更します。たとえば、
toをftoに、fromをffromに変更します。 -
予約キーワードを使用するフィールドをバッククォート () で囲みます。
-
エラー: "Invalid primary key. Column 'xxx' is nullable."
-
原因
Flink では、すべてのプライマリキーカラムを
NOT NULLとして明示的に宣言する必要があります。データに NULL 値が含まれていない場合でも、テーブル作成ステートメントでプライマリキーカラムが NULL 値を許容する場合 (たとえば、INT NULL)、 Flink は書き込み前に操作を拒否します。これは実行時エラーではなく、DDL 解析フェーズでのセマンティックチェックです。 -
解決策
エラーメッセージで指摘されているプライマリキーカラムを
NOT NULLとして宣言し、テーブルを再作成します。
JSON ファイルがダウンロードされずにブラウザで開かれる
-
症状
Artifacts ページから JSON ファイルをダウンロードしようとクリックしても、ブラウザはダウンロードをトリガーしません。代わりに、新しいタブを開いて JSON コンテンツを直接表示します。
-
原因
OSS 内の JSON ファイルに
Content-Disposition: attachmentHTTP レスポンスヘッダーがありません。これにより、ブラウザはファイルをダウンロードする代わりに、ファイルのコンテンツを直接表示します。 -
解決策
-
オプション 1: ファイルを再アップロードする
この問題はプラットフォームバージョン 4.5.0 で修正されていますが、修正はこのバージョンのリリース後にアップロードされたファイルにのみ適用されます。この日付より前にアップロードされたファイルは、手動で処理する必要があります。
-
オプション 2: OSS オブジェクトメタデータを変更する
次の標準 HTTP 属性を追加して、オブジェクトメタデータを手動で変更します。
-
ヘッダー名： Content-Disposition
-
ヘッダー値： attachment
詳細については、「オブジェクトメタデータの管理」をご参照ください。
-
-