Development and debugging - Realtime Compute for Apache Flink

Bu konu, yaygın geliştirme ve hata ayıklama sorunlarını kapsar.

DDL ifadesini aynı betikte DML ifadeleriyle birlikte gönderirken nasıl bildirmeliyim?
Birden fazla INSERT INTO ifadesini nasıl yazarım?
Entry point Main Arguments içinde özel karakterleri nasıl iletirim?
Birden fazla değişiklikten sonra UDF JAR paketinin yüklenmesi neden başarısız olur?
POJO sınıfı UDTF dönüş türü olarak kullanıldığında alanlar neden hizalanmaz?
Flink bağımlılık çakışmalarını nasıl çözerim?
Hata: Could not parse type at position 50: expected but was . Input type string: ROW
Tabloya yazma sırasında hata: "Invalid primary key. Column 'xxx' is nullable."
JSON dosyası indirilmek yerine tarayıcıda neden açılır?

DDL ifadesini DML ifadeleriyle birlikte bildirme

Aynı betik içinde DDL ve DML ifadelerini birlikte gönderdiğinizde, DDL ifadesini CREATE TABLE yerine CREATE TEMPORARY TABLE olarak bildirin. Aksi takdirde, Validate seçeneğine tıklamak aşağıdakine benzer bir hatayla başarısız olur.

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;
Error message:
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)

Birden fazla INSERT INTO ifadesi

Tek bir mantıksal birim oluşturmak için birden fazla INSERT INTO ifadesini BEGIN STATEMENT SET; ve END; arasında sarmalayın. Daha fazla bilgi için bkz. INSERT INTO ifadeleri. İfadeleri sarmazsanız, Validate seçeneğine tıklamak aşağıdakine benzer bir hatayla başarısız olur.

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)

Ana argümanlarda özel karakterleri iletme

Neden

Entry point Main Arguments içinde # ve $ gibi özel karakterleri ilettiğinizde, ters eğik çizgi (\) kaçış karakteri çalışmaz ve özel karakterler yok sayılır.
Çözüm

Deployments sayfasında hedef dağıtımın adına tıklayın. Parameters bölümünde, env.java.opts: -Dconfig.disable-inline-comment=true parametresini Other Configuration alanına ekleyin. Daha fazla bilgi için bkz. Özel dağıtım parametrelerini yapılandırma.

Değişiklikten sonra UDF JAR yükleme hatası

Neden

UDF çalışma zamanı, JAR paketleri arasında benzersiz sınıf adları zorunlu kılar.
Çözüm
- Eski JAR paketini silin ve yeni olanı yükleyin.
- JAR paketini Additional Dependencies bölümünde yükleyin ve kodunuzda geçici bir işlev kullanın. Geçici işlev kullanımı hakkında bilgi için bkz. UDF kaydetme. Aşağıdaki örnek sözdizimini gösterir.
```
CREATE TEMPORARY FUNCTION `cp_record_reduce` AS 'com.taobao.test.udf.blink.CPRecordReduceUDF';
```
  Sağ taraftaki Other Configuration panelinde bulunan Additional Dependencies bölümünde, UDF JAR paketinizin OSS URL'sini sağlayın.

UDTF dönüş türü olarak POJO sınıfı kullanıldığında alan hizalama sorunu

Ayrıntılar

Bir UDTF için dönüş türü olarak bir POJO sınıfı kullandığınızda ve döndürülen sütunlar için SQL'de açıkça bir takma ad listesi bildirdiğinizde, alan hizalama sorunuyla karşılaşabilirsiniz. Sonuç olarak, veri türleri tutarlı olsa bile gerçek alanlar beklediğiniz alanlar olmayabilir.

Örneğin, aşağıdaki POJO sınıfını bir UDTF için dönüş türü olarak kullanıp paketlerseniz ve işlevi Dağıtım düzeyinde UDF olarak Özel işlev geliştirme bölümünde açıklandığı gibi kaydederseniz, SQL doğrulaması başarısız olur.

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 doğrulaması aşağıdaki hata iletisini döndürür:

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'den döndürülen alanlar, POJO sınıfındaki alanlarla hizalanmaz. Sorgu sonucunda, c alanı BOOLEAN türündedir ve a alanı INT türündedir; bu, POJO sınıfındaki tanımlarının tersidir.

Neden

POJO sınıfları için tür çıkarım kurallarına göre:
- POJO sınıfının parametreli bir yapıcısı varsa, Flink dönüş türünü yapıcının parametre sırasına göre çıkarır.
- POJO sınıfının parametreli bir yapıcısı yoksa, Flink alanları ada göre alfabetik sıraya göre yeniden sıralar.
Örnekte, UDTF'nin dönüş türü için kullanılan POJO sınıfı parametreli bir yapıcıya sahip olmadığından, alanlar alfabetik sıraya göre döndürülür ve sonuçta BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d) türü oluşur. Bu çıkarım geçerli olsa da, SQL sorgusu çıktı sütunlarını LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b) ile açıkça yeniden adlandırır. Bu takma ad listesi, sütunları konuma göre yeniden adlandırır ve POJO'dan gelen alfabetik olarak sıralanmış alanlarla uyumsuzluk yaratır. Konumsal takma adlandırma ile alfabetik alan sıralaması arasındaki bu çakışma, doğrulama istisnasına veya beklenmeyen veri hizalama sorununa neden olur.

Çözüm

POJO sınıfı parametreli bir yapıcıdan yoksunsa, UDTF dönüş alanlarının açık yeniden adlandırılmasını kaldırın. Örneğin, SQL'deki INSERT ifadesini şu şekilde değiştirin:

-- If the POJO class lacks a parameterized constructor, select the required fields by name. 
-- When using T.*, you must be aware of the actual order of the returned fields.
SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T;

Dönüş türündeki alanların sırasını kontrol etmek için POJO sınıfında parametreli bir yapıcı uygulayın. Bu durumda, UDTF çıktısının alan sırası, yapıcının parametre sırasıyla eşleşir.

package com.aliyun.example;
public class TestPojoWithConstructor {
	public int c;
	public String d;
	public boolean a;
	public String b;
	// Using specific fields order instead of alphabetical order
	public TestPojoWithConstructor(int c, String d, boolean a, String b) {
		this.c = c;
		this.d = d;
		this.a = a;
		this.b = b;
	}
}

Flink bağımlılık çakışmalarını çözme

Belirtiler
- Çakışma, Flink veya Hadoop ile ilgili sınıflar tarafından oluşturulan açık hatalar olarak kendini gösterir.
```
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
```
- Alternatif olarak, sorun açık bir hata iletisi olmadan beklenmeyen davranış olarak ortaya çıkabilir, örneğin:
  - Günlükler oluşturulmaz veya log4j yapılandırması etkili olmaz.
    
    Bu sorun genellikle bağımlılıklarda bulunan log4j ile ilgili yapılandırmalardan kaynaklanır. Dağıtım JAR paketinin log4j yapılandırmaları taşıyan bağımlılıklar içerip içermediğini kontrol edin. Bağımlılık tanımlarınızda dışlamalar kullanarak bu yapılandırmaları kaldırabilirsiniz.
    
    Not
    Farklı bir log4j sürümü kullanmanız gerekiyorsa, log4j ile ilgili sınıfları yeniden konumlandırmak için maven-shade-plugin kullanın.
  - RPC çağrı istisnaları.
    
    Flink'in Akka RPC çağrılarını etkileyen bağımlılık çakışmaları, varsayılan olarak günlüklerde görüntülenmeyen istisnalara neden olabilir. Bunları belirlemek için hata ayıklama günlüğünü etkinleştirmeniz gerekir.
    
    Örneğin, hata ayıklama günlüğü Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx} gösterir, ancak JobManager (JM) günlüğü Registering TaskManager with ResourceID xxx sonrasında NoResourceAvailableException zaman aşımı hatası oluşana kadar herhangi bir etkinlik göstermez. Bu arada, TaskManager (TM) sürekli olarak Cannot allocate the requested resources. Trying to allocate ResourceProfile{xxx} hatasını bildirir.
    
    Neden: Hata ayıklama günlüğü etkinleştirildiğinde, bir RPC çağrısı sırasında bir InvocationTargetException oluşturulduğunu görebilirsiniz. Bu hata, TM yuva tahsisinin işlem ortasında başarısız olmasına ve tutarsız bir duruma neden olur. ResourceManager (RM) daha sonra sürekli olarak ve başarısız bir şekilde bir yuva tahsis etmeye çalışır ve kurtaramaz.
Nedenler
- Dağıtım JAR paketi, temel Flink, Hadoop veya log4j kitaplıkları gibi gereksiz bağımlılıklar içerir ve bu durum bağımlılık çakışmalarına neden olur.
- Gerekli bir konektör için bağımlılıklar JAR paketinde yer almaz.
Sorun giderme
- Dağıtımın pom.xml dosyasını gereksiz bağımlılıklar için inceleyin.
- Çakışan dosyaları kontrol etmek için jar tf foo.jar komutunu çalıştırarak dağıtım JAR paketinin içeriğini denetleyin.
- Çakışmaları analiz etmek için dağıtımın bağımlılık ağacını mvn dependency:tree komutunu çalıştırarak inceleyin.

Çözüm

En iyi uygulama olarak, temel çerçeve bağımlılıklarının scope değerini provided olarak ayarlayın. Bu, bunların dağıtım JAR paketine dahil edilmesini önler.

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>

Gerekli konektör bağımlılıklarını projenize ekleyin. Varsayılan kapsam compile olup, bunları dağıtım JAR paketine doğru şekilde dahil eder. Örneğin, Kafka konektörünü eklemek için:
```
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
```
Diğer Flink, Hadoop veya log4j bağımlılıklarını eklemeyin. Ancak:
- Dağıtımın temel yapılandırma veya konektörle ilgili bileşenlere doğrudan bağımlılığı varsa, kapsamı provided olarak ayarlayın. Aşağıdaki örnek sözdizimini gösterir.
```
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <scope>provided</scope>
</dependency>
```
- Dağıtımın temel yapılandırma veya konektörle ilgili bileşenlere geçişli bağımlılığı varsa, bir exclusion kullanarak bunu kaldırın. Aşağıdaki örnek sözdizimini gösterir.
```
<dependency>
    <groupId>foo</groupId>
      <artifactId>bar</artifactId>
      <exclusions>
        <exclusion>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
       </exclusion>
    </exclusions>
</dependency>
```

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

Hata ayrıntıları

SQL düzenleyicide SQL yazarken, bir UDTF kullandığınızda bir sözdizimi kontrol hatası (kırmızı dalgalı çizgi) oluşur.

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>

Aşağıdaki kod bir örnektir:

@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);
    }
}

Neden

İşlev veri türlerini tanımlamak için DataTypeHint kullandığınızda, ayrılmış bir anahtar kelime doğrudan alan adı olarak kullanılır.
Çözüm
- Alanları anahtar kelime olmayan adlarla yeniden adlandırın. Örneğin, to öğesini fto ve from öğesini ffrom olarak yeniden adlandırın.
- Ayrılmış anahtar kelimeler kullanan alanları geri tırnak içine alın (`).

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

Neden

Flink, tüm birincil anahtar sütunlarının açıkça NOT NULL olarak bildirilmesini zorunlu kılar. Veriler NULL değer içermese bile, tablo oluşturma ifadesindeki bir birincil anahtar sütunu NULL değerlere izin veriyorsa (örneğin, INT NULL), Flink yazma işleminden önce işlemi reddeder. Bu bir çalışma zamanı hatası değil, DDL ayrıştırma aşamasındaki anlamsal bir kontroldür.
Çözüm

Hatada belirtilen birincil anahtar sütunlarını NOT NULL olarak bildirin ve tabloyu yeniden oluşturun.

JSON dosyası indirilmek yerine tarayıcıda açılır

Belirti

Artifacts sayfasından bir JSON dosyasını indirmek için tıkladığınızda, tarayıcı indirme işlemini tetiklemez. Bunun yerine, yeni bir sekme açar ve JSON içeriğini doğrudan görüntüler.
Neden

OSS'teki JSON dosyasında Content-Disposition: attachment HTTP yanıt üstbilgisi eksiktir. Bu durum, tarayıcının dosyayı indirmek yerine içeriğini doğrudan görüntülemesine neden olur.
Çözüm
- Seçenek 1: Dosyayı yeniden yükleme
  
  Bu sorun platform sürüm 4.5.0'da düzeltilmiştir, ancak düzeltme yalnızca bu sürüm yayınlandıktan sonra yüklenen dosyalar için geçerlidir. Bu tarihten önce yüklenen dosyalar manuel olarak işlenmelidir.
- Seçenek 2: OSS nesne meta verilerini değiştirme
  
  Aşağıdaki standart HTTP özniteliğini ekleyerek nesne meta verilerini manuel olarak değiştirin:
  - Üstbilgi adı: Content-Disposition
  - Üstbilgi değeri: attachment
  Daha fazla bilgi için bkz. Nesne meta verilerini yönetme.