SQL Client
SQL Client
Flink 의 Table & SQL API 는 SQL 언어로 작성된 쿼리로 작업하는 것을 가능하게 하지만, 이러한 쿼리는 Java 나 Scala 로 작성된 테이블 프로그램 안에 포함되어야 합니다. 또한 이러한 프로그램은 클러스터에 제출되기 전에 빌드 도구로 패키징되어야 합니다. 이는 어느 정도 Flink 의 사용을 Java/Scala 프로그래머로 제한합니다.
SQL Client 는 Java 나 Scala 코드 한 줄 없이 테이블 프로그램을 작성·디버깅·제출하는 쉬운 방법을 제공하는 것을 목표로 합니다. SQL Client CLI 는 명령줄에서 실행 중인 분산 애플리케이션의 실시간 결과를 검색하고 시각화하는 것을 허용합니다.
출처: 문서
본문
Flink 의 Table & SQL API 는 SQL 언어로 작성된 쿼리로 작업하는 것을 가능하게 하지만, 이러한 쿼리는 Java 나 Scala 로 작성된 테이블 프로그램 안에 포함되어야 합니다. 또한 이러한 프로그램은 클러스터에 제출되기 전에 빌드 도구로 패키징되어야 합니다. 이는 어느 정도 Flink 의 사용을 Java/Scala 프로그래머로 제한합니다.
SQL Client 는 Java 나 Scala 코드 한 줄 없이 테이블 프로그램을 작성·디버깅·제출하는 쉬운 방법을 제공하는 것을 목표로 합니다. SQL Client CLI 는 명령줄에서 실행 중인 분산 애플리케이션의 실시간 결과를 검색하고 시각화하는 것을 허용합니다.
시작하기 (Getting Started)
이 섹션에서는 명령줄에서 첫 Flink SQL 프로그램을 설정하고 실행하는 방법을 설명합니다.
SQL Client 는 일반 Flink 배포에 번들되어 있어 바로 실행할 수 있습니다. 테이블 프로그램이 실행될 수 있는 실행 중인 Flink 클러스터만 필요합니다. Flink 클러스터 설정에 대한 자세한 내용은 Cluster & Deployment 부분을 참고하세요. SQL Client 를 시험해 보고 싶다면 다음 명령으로 worker 하나가 있는 로컬 클러스터를 시작할 수도 있습니다:
./bin/start-cluster.sh
SQL Client CLI 시작하기
SQL Client 스크립트는 Flink 의 바이너리 디렉터리에도 있습니다. 사용자는 SQL Client CLI 를 시작하는 두 가지 옵션이 있습니다. 임베디드(embedded) 독립 실행 프로세스를 시작하거나 원격 SQL Gateway 에 연결하는 것입니다. SQL Client 기본 모드는 embedded 입니다.
임베디드 모드에서 CLI 를 시작할 수 있습니다:
./bin/sql-client.sh
또는 명시적으로 embedded 모드를 사용합니다:
./bin/sql-client.sh embedded
게이트웨이 모드의 경우 CLI 를 다음으로 시작할 수 있습니다:
./bin/sql-client.sh gateway --endpoint <gateway address>
게이트웨이 모드에서 CLI 는 명령문을 실행하기 위해 SQL 을 지정된 원격 게이트웨이에 제출합니다.
<gateway address> 는 host:port 조합 또는 전체 URL 의 두 가지 형식으로 제공될 수 있습니다.
클라이언트를 초기화하면서 클러스터 주소를 명시적으로 제공해 SQL client 의 기본 job manager 구성을 재정의할 수도 있습니다:
./bin/sql-client.sh gateway --endpoint <gateway address> -Drest.address=<flink cluster host> -Drest.port=<port>
사용자 정의 HTTP 헤더를 전달해야 한다면 FLINK_REST_CLIENT_HEADERS 환경 변수를 설정하면 됩니다. 예:
export FLINK_REST_CLIENT_HEADERS="Cookie:myauthcookie=foobar;othercookie=baz"
./bin/sql-client.sh gateway --endpoint https://your-sql-gateway.endpoint.com/authenticated/sql
여러 헤더의 경우 새 줄로 구분합니다:
export FLINK_REST_CLIENT_HEADERS=$(cat << EOF
Cookie:myauthcookie=foobar
Cache-Control: no-cache
EOF)
기본적으로 SQL Client 는 SQL client 쪽의 Flink 구성 파일 에서 security.ssl.rest.truststore 및 security.ssl.rest.truststore-password 속성으로 구성된 truststore 를 사용합니다. 이 속성이 명시적으로 구성되지 않으면 클라이언트는 JDK 가 제공하는 기본 인증서 저장소를 사용합니다.
참고: SQL Client 는 버전 v2 이후부터 REST Endpoint 에 연결하는 것만 지원합니다.
자세한 내용은 아래의 SQL Client 시작 옵션 을 참고하세요.
SQL 쿼리 실행하기
설정과 클러스터 연결을 검증하려면 아래의 간단한 쿼리를 입력하고 Enter 를 눌러 실행할 수 있습니다:
SET 'sql-client.execution.result-mode' = 'tableau';
SET 'execution.runtime-mode' = 'batch';
SELECT
name,
COUNT(*) AS cnt
FROM
(VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name)
GROUP BY name;
SQL Client 는 클러스터에서 결과를 검색해 시각화합니다(Q 키를 눌러 결과 뷰를 닫을 수 있습니다):
+-------+-----+
| name | cnt |
+-------+-----+
| Alice | 1 |
| Bob | 2 |
| Greg | 1 |
+-------+-----+
SET 명령은 작업 실행과 sql client 동작을 조정할 수 있게 해줍니다. 자세한 내용은 아래의 SQL Client 구성 을 참고하세요.
쿼리가 정의된 후 장기 실행되는 분리형(detached) Flink 작업으로 클러스터에 제출될 수 있습니다. 구성 섹션 은 데이터 읽기를 위한 테이블 소스를 선언하는 방법, 데이터 쓰기를 위한 테이블 sink 를 선언하는 방법, 기타 테이블 프로그램 속성을 구성하는 방법을 설명합니다.
키 입력 (Key-strokes)
SQL Client 에 사용할 수 있는 키 입력 목록이 있습니다:
| 키 입력 (Linux, Windows(WSL)) | 키 입력 (Mac) | 설명 |
|---|---|---|
alt-b, ctrl+⍇ |
Esc-b |
Backward word |
alt-f, Ctrl+⍈ |
Esc-f |
Forward word |
alt-c |
Esc-c |
Capitalize word |
alt-l |
Esc-l |
Lowercase word |
alt-u |
Esc-u |
Uppercase word |
alt-d |
Esc-d |
Kill word |
alt-n |
Esc-n |
History search forward (빈 입력이면 history 에서 다음 줄과 동일) |
alt-p |
Esc-p |
History search backward (빈 입력이면 history 에서 이전 줄과 동일) |
alt-t |
Esc-t |
Transpose words |
ctrl-a |
⌘-a |
줄의 시작으로 |
ctrl-e |
⌘-e |
줄의 끝으로 |
ctrl-b |
⌘-b |
Backward char |
ctrl-f |
⌘-f |
Forward char |
ctrl-d |
⌘-d |
Delete char |
ctrl-h |
⌘-h |
Backward delete char |
ctrl-t |
⌘-t |
Transpose chars |
ctrl-i |
⌘-i |
완성(completion) 호출 |
ctrl-j |
⌘-j |
쿼리 제출 |
ctrl-m |
⌘-m |
쿼리 제출 |
ctrl-k |
⌘-k |
커서 오른쪽 줄을 죽임 |
ctrl-w |
⌘-w |
커서 왼쪽 줄을 죽임 |
ctrl-u |
⌘-u |
전체 줄을 죽임 |
ctrl-l |
⌘-l |
화면 지우기 |
ctrl-n |
⌘-n |
history 에서 아래 줄 |
ctrl-p |
⌘-p |
history 에서 위 줄 |
ctrl-r |
⌘-r |
History 증분 검색(뒤로) |
ctrl-s |
⌘-s |
History 증분 검색(앞으로) |
도움말 얻기
SQL Client 명령의 문서는 HELP 명령을 입력해 접근할 수 있습니다.
일반적인 SQL 문서도 참고하세요.
구성 (Configuration)
SQL Client 시작 옵션
SQL Client 는 다음 선택적 CLI 명령으로 시작할 수 있습니다. 이들은 다음 단락에서 자세히 논의됩니다:
./sql-client [MODE] [OPTIONS]
The following options are available:
Mode "embedded" (default) submits Flink jobs from the local machine.
Syntax: [embedded] [OPTIONS]
"embedded" mode options:
-D <session dynamic config key=val> The dynamic config key=val for a
session.
-f,--file <script file> Script file that should be
executed. In this mode, the
client will not open an
interactive terminal.
-h,--help Show the help message with
descriptions of all options.
-hist,--history <History file path> The file which you want to save
the command history into. If not
specified, we will auto-generate
one under your user's home
directory.
-i,--init <initialization file> Script file that used to init
the session context. If get
error in execution, the sql
client will exit. Notice it's
not allowed to add query or
insert into the init file.
-j,--jar <JAR file> A JAR file to be imported into
the session. The file might
contain user-defined classes
needed for the execution of
statements such as functions,
table sources, or sinks. Can be
used multiple times.
-l,--library <JAR directory> A JAR file directory with which
every new session is
initialized. The files might
contain user-defined classes
needed for the execution of
statements such as functions,
table sources, or sinks. Can be
used multiple times.
-pyarch,--pyArchives <arg> Add python archive files for
job. The archive files will be
extracted to the working
directory of python UDF worker.
For each archive file, a target
directory be specified. If the
target directory name is
specified, the archive file will
be extracted to a directory with
the specified name. Otherwise,
the archive file will be
extracted to a directory with
the same name of the archive
file. The files uploaded via
this option are accessible via
relative path. '#' could be used
as the separator of the archive
file path and the target
directory name. Comma (',')
could be used as the separator
to specify multiple archive
files. This option can be used
to upload the virtual
environment, the data files used
in Python UDF (e.g.,
--pyArchives
file:///tmp/py37.zip,file:///tmp
/data.zip#data --pyExecutable
py37.zip/py37/bin/python). The
data files could be accessed in
Python UDF, e.g.: f =
open('data/data.txt', 'r').
-pyclientexec,--pyClientExecutable <arg> The path of the Python
interpreter used to launch the
Python process when submitting
the Python jobs via "flink run"
or compiling the Java/Scala jobs
containing Python UDFs.
-pyexec,--pyExecutable <arg> Specify the path of the python
interpreter used to execute the
python UDF worker (e.g.:
--pyExecutable
/usr/local/bin/python3). The
python UDF worker depends on
Python 3.9+, Apache Beam
(version >= 2.54.0, <= 2.61.0), Pip
(version >= 20.3) and SetupTools
(version >= 37.0.0). Please
ensure that the specified
environment meets the above
requirements.
-pyfs,--pyFiles <pythonFiles> Attach custom files for job. The
standard resource file suffixes
such as .py/.egg/.zip/.whl or
directory are all supported.
These files will be added to the
PYTHONPATH of both the local
client and the remote python UDF
worker. Files suffixed with .zip
will be extracted and added to
PYTHONPATH. Comma (',') could be
used as the separator to specify
multiple files (e.g., --pyFiles
file:///tmp/myresource.zip,hdfs:
///$namenode_address/myresource2
.zip).
-pyreq,--pyRequirements <arg> Specify a requirements.txt file
which defines the third-party
dependencies. These dependencies
will be installed and added to
the PYTHONPATH of the python UDF
worker. A directory which
contains the installation
packages of these dependencies
could be specified optionally.
Use '#' as the separator if the
optional parameter exists (e.g.,
--pyRequirements
file:///tmp/requirements.txt#fil
e:///tmp/cached_dir).
-s,--session <session identifier> The identifier for a session.
'default' is the default
identifier.
Mode "gateway" mode connects to the SQL gateway for submission.
Syntax: gateway [OPTIONS]
"gateway" mode options:
-D <session dynamic config key=val> The dynamic config key=val for a
session.
-e,--endpoint <SQL Gateway address> The address of the remote SQL Gateway
to connect.
-f,--file <script file> Script file that should be executed.
In this mode, the client will not
open an interactive terminal.
-h,--help Show the help message with
descriptions of all options.
-hist,--history <History file path> The file which you want to save the
command history into. If not
specified, we will auto-generate one
under your user's home directory.
-i,--init <initialization file> Script file that used to init the
session context. If get error in
execution, the sql client will exit.
Notice it's not allowed to add query
or insert into the init file.
-s,--session <session identifier> The identifier for a session.
'default' is the default identifier.
SQL Client 구성
아래 옵션을 설정하거나 유효한 Flink 구성 항목을 설정해 SQL Client 를 구성할 수 있습니다:
SET 'key' = 'value';
| 키 | 기본값 | 유형 | 설명 |
|---|---|---|---|
sql-client.display.color-schema (Batch Streaming) |
"DEFAULT" | String | SQL client 에서 사용할 SQL 하이라이트 색상 스키마. 가능한 값: 'default', 'dark', 'light', 'chester', 'vs2010', 'solarized', 'obsidian', 'geshi'. |
sql-client.display.print-time-cost (Batch) |
true | Boolean | 쿼리의 시간 소비 표시 여부를 결정합니다. 기본적으로 쿼리 시간 비용은 표시되지 않습니다. |
sql-client.display.show-line-numbers (Batch Streaming) |
false | Boolean | 다중 라인 SQL 에 줄 번호를 표시할지 여부를 결정합니다. |
sql-client.execution.max-table-result.rows (Batch Streaming) |
1000000 | Integer | 테이블 모드에서 캐시할 행 수. 행 수가 지정된 값을 초과하면 FIFO 방식으로 행을 재시도합니다. |
sql-client.execution.result-mode (Batch Streaming) |
TABLE | Enum | 쿼리 결과가 표시되는 방식을 결정합니다. 가능한 값: "TABLE"(결과를 메모리에 구체화하고 일반적인 페이지 매김 테이블 표현으로 시각화), "CHANGELOG"(지속 쿼리가 생성한 결과 스트림을 시각화), "TABLEAU"(결과를 tableau 형식으로 화면에 직접 표시). |
sql-client.verbose (Batch Streaming) |
false | Boolean | 콘솔에 verbose 출력을 출력할지 여부를 결정합니다. true 로 설정하면 예외 스택을 출력합니다. 그렇지 않으면 원인만 출력합니다. |
SQL Client 결과 모드
CLI 는 결과를 유지하고 시각화하는 세 가지 모드 를 지원합니다.
table 모드 는 결과를 메모리에 구체화하고 일반적인 페이지 매김 테이블 표현으로 시각화합니다. CLI 에서 다음 명령을 실행해 활성화할 수 있습니다:
SET 'sql-client.execution.result-mode' = 'table';
쿼리 결과는 그런 다음 다음과 같이 보이며, 화면 하단에 표시된 키와 화살표 키를 사용해 다양한 레코드를 탐색하고 열 수 있습니다:
name age isHappy dob height
user1 20 true 1995-12-03 1.7
user2 30 true 1972-08-02 1.89
user3 40 false 1983-12-23 1.63
user4 41 true 1977-11-13 1.72
user5 22 false 1998-02-20 1.61
user6 12 true 1969-04-08 1.58
user7 38 false 1987-12-15 1.6
user8 62 true 1996-08-05 1.82
Q Quit + Inc Refresh G Goto Page N Next Page O Open Row
R Refresh - Dec Refresh L Last Page P Prev Page
changelog 모드 는 결과를 구체화하지 않고 삽입(+)과 재시도(-)로 구성된 지속 쿼리 가 생성한 결과 스트림을 시각화합니다.
SET 'sql-client.execution.result-mode' = 'changelog';
쿼리 결과는 그런 다음 다음과 같이 보입니다:
op name age isHappy dob height
+I user1 20 true 1995-12-03 1.7
+I user2 30 true 1972-08-02 1.89
+I user3 40 false 1983-12-23 1.63
+I user4 41 true 1977-11-13 1.72
+I user5 22 false 1998-02-20 1.61
+I user6 12 true 1969-04-08 1.58
+I user7 38 false 1987-12-15 1.6
+I user8 62 true 1996-08-05 1.82
Q Quit + Inc Refresh O Open Row
R Refresh - Dec Refresh
tableau 모드 는 더 전통적인 방식으로 결과를 tableau 형식으로 화면에 직접 표시합니다. 표시 내용은 쿼리 실행 유형(execution.type)의 영향을 받습니다.
SET 'sql-client.execution.result-mode' = 'tableau';
쿼리 결과는 그런 다음 다음과 같이 보입니다:
+----+--------------------------------+-------------+---------+------------+--------------------------------+
| op | name | age | isHappy | dob | height |
+----+--------------------------------+-------------+---------+------------+--------------------------------+
| +I | user1 | 20 | true | 1995-12-03 | 1.7 |
| +I | user2 | 30 | true | 1972-08-02 | 1.89 |
| +I | user3 | 40 | false | 1983-12-23 | 1.63 |
| +I | user4 | 41 | true | 1977-11-13 | 1.72 |
| +I | user5 | 22 | false | 1998-02-20 | 1.61 |
| +I | user6 | 12 | true | 1969-04-08 | 1.58 |
| +I | user7 | 38 | false | 1987-12-15 | 1.6 |
| +I | user8 | 62 | true | 1996-08-05 | 1.82 |
+----+--------------------------------+-------------+---------+------------+--------------------------------+
Received a total of 8 rows
이 모드를 스트리밍 쿼리와 함께 사용하면 결과가 콘솔에 지속적으로 인쇄됩니다. 이 쿼리의 입력 데이터가 유한하면 Flink 가 모든 입력 데이터를 처리한 후 작업이 종료되고 인쇄도 자동으로 중지됩니다. 그렇지 않으면 실행 중인 쿼리를 종료하려면 CTRL-C 를 입력하면 작업과 인쇄가 중지됩니다.
이 모든 결과 모드는 SQL 쿼리 프로토타이핑 중에 유용합니다. 모든 모드에서 결과는 SQL Client 의 Java 힙 메모리에 저장됩니다. CLI 인터페이스의 응답성을 유지하기 위해 changelog 모드는 최신 1000 개의 변경만 표시합니다. table 모드는 사용 가능한 주 메모리와 구성된 [최대 행 수(sql-client.execution.max-table-result.rows)] 로만 제한되는 더 큰 결과를 탐색할 수 있습니다.
주의: 배치 환경에서 실행되는 쿼리는
table또는tableau결과 모드로만 검색할 수 있습니다.
SQL 파일로 세션 초기화
SQL 쿼리는 실행되는 구성 환경이 필요합니다. SQL Client 는 SQL Client 가 시작될 때 환경을 설정하기 위해 초기화 SQL 파일을 실행하는 -i 시작 옵션을 지원합니다. 소위 초기화 SQL 파일 은 사용 가능한 카탈로그, 테이블 소스와 sink, 사용자 정의 함수, 실행 및 배포에 필요한 기타 속성을 정의하는 DDL 을 사용할 수 있습니다.
그러한 파일의 예는 아래에 제시됩니다:
-- Define available catalogs
CREATE CATALOG MyCatalog
WITH (
'type' = 'hive'
);
USE CATALOG MyCatalog;
-- Define available database
CREATE DATABASE MyDatabase;
USE MyDatabase;
-- Define TABLE
CREATE TABLE MyTable(
MyField1 INT,
MyField2 STRING
) WITH (
'connector' = 'filesystem',
'path' = '/path/to/something',
'format' = 'csv'
);
-- Define VIEW
CREATE VIEW MyCustomView AS SELECT MyField2 FROM MyTable;
-- Define user-defined functions here.
CREATE FUNCTION myUDF AS 'foo.bar.AggregateUDF';
-- Properties that change the fundamental execution behavior of a table program.
SET 'execution.runtime-mode' = 'streaming'; -- execution mode either 'batch' or 'streaming'
SET 'sql-client.execution.result-mode' = 'table'; -- available values: 'table', 'changelog' and 'tableau'
SET 'sql-client.execution.max-table-result.rows' = '10000'; -- optional: maximum number of maintained rows
SET 'parallelism.default' = '1'; -- optional: Flink's parallelism (1 by default)
SET 'pipeline.auto-watermark-interval' = '200'; --optional: interval for periodic watermarks
SET 'pipeline.max-parallelism' = '10'; -- optional: Flink's maximum parallelism
SET 'table.exec.state.ttl' = '1000'; -- optional: table program's idle state time
SET 'restart-strategy.type' = 'fixed-delay';
-- Configuration options for adjusting and tuning table programs.
SET 'table.optimizer.join-reorder-enabled' = 'true';
SET 'table.exec.spill-compression.enabled' = 'true';
SET 'table.exec.spill-compression.block-size' = '128kb';
이 구성은:
- Hive 카탈로그에 연결하고
MyCatalog를 현재 카탈로그로,MyDatabase를 카탈로그의 현재 데이터베이스로 사용하며, - CSV 파일에서 데이터를 읽을 수 있는 테이블
MyTable을 정의하고, - SQL 쿼리를 사용해 가상 테이블을 선언하는 뷰
MyCustomView를 정의하며, - 클래스 이름으로 인스턴스화할 수 있는 사용자 정의 함수
myUDF를 정의하고, - 명령문 실행에 스트리밍 모드와 병렬도 1을 사용하며,
table결과 모드에서 탐색 쿼리를 실행하고,- 구성 옵션을 통해 join 재정렬과 spilling 관련 일부 플래너 조정을 수행합니다.
-i <init.sql> 옵션으로 SQL Client 세션을 초기화할 때 초기화 SQL 파일에서 다음 문이 허용됩니다:
- DDL(CREATE/DROP/ALTER),
- USE CATALOG/DATABASE,
- LOAD/UNLOAD MODULE,
- SET 명령,
- RESET 명령.
쿼리 또는 insert 문을 실행할 때는 대화형 모드로 들어가거나 -f 옵션을 사용해 SQL 문을 제출하세요.
주의: 초기화 중 SQL Client 가 오류를 받으면 SQL Client 는 오류 메시지와 함께 종료됩니다.
의존성 (Dependencies)
SQL Client 는 Maven, Gradle, sbt 를 사용한 Java 프로젝트 설정을 요구하지 않습니다. 대신 의존성을 클러스터에 제출되는 일반 JAR 파일로 전달할 수 있습니다. 각 JAR 파일을 개별적으로 지정하거나(--jar) 전체 라이브러리 디렉터리를 정의할 수 있습니다(--library). 외부 시스템(예: Apache Kafka)용 커넥터와 그에 상응하는 데이터 포맷(예: JSON)을 위해 Flink 는 즉시 사용 가능한 JAR 번들 을 제공합니다. 이 JAR 파일은 각 릴리스에 대해 Maven 중앙 저장소에서 다운로드할 수 있습니다.
제공되는 SQL JAR 의 전체 목록은 외부 시스템 연결 페이지 에서 찾을 수 있습니다.
커넥터 및 포맷 의존성을 구성하는 방법에 대한 정보는 구성 섹션을 참고하세요.
사용법 (Usage)
SQL Client 는 사용자가 대화형 명령줄 내에서 작업을 제출하거나 -f 옵션을 사용해 SQL 파일을 실행할 수 있게 해줍니다.
두 모드 모두 SQL Client 는 Flink 가 지원하는 모든 유형의 SQL 문을 파싱하고 실행하는 것을 지원합니다.
대화형 명령줄 (Interactive Command Line)
대화형 명령줄에서 SQL Client 는 사용자 입력을 읽고 세미콜론(;)으로 종료되는 문을 실행합니다.
SQL Client 는 문이 성공적으로 실행되면 성공 메시지를 인쇄합니다. 오류가 발생하면 SQL Client 도 오류 메시지를 인쇄합니다. 기본적으로 오류 메시지는 오류 원인만 포함합니다. 디버깅을 위해 전체 예외 스택을 인쇄하려면 SET 'sql-client.verbose' = 'true'; 명령을 통해 sql-client.verbose 를 true 로 설정하세요.
세션 클러스터에서 SQL 파일 실행
SQL Client 는 -f 옵션으로 SQL 스크립트 파일 실행을 지원합니다. SQL Client 는 SQL 스크립트 파일의 문을 하나씩 실행하고 각 실행된 문에 대한 실행 메시지를 인쇄합니다. 문이 실패하면 SQL Client 는 종료되고 나머지 문은 실행되지 않습니다.
그러한 파일의 예는 아래에 제시됩니다:
CREATE TEMPORARY TABLE users (
user_id BIGINT,
user_name STRING,
user_level STRING,
region STRING,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'users',
'properties.bootstrap.servers' = '...',
'key.format' = 'csv',
'value.format' = 'avro'
);
-- set sync mode
SET 'table.dml-sync' = 'true';
-- set the job name
SET 'pipeline.name' = 'SqlJob';
-- set the queue that the job submit to
SET 'yarn.application.queue' = 'root';
-- set the job parallelism
SET 'parallelism.default' = '100';
-- restore from the specific savepoint path
SET 'execution.state-recovery.path' = '/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab';
INSERT INTO pageviews_enriched
SELECT *
FROM pageviews AS p
LEFT JOIN users FOR SYSTEM_TIME AS OF p.proctime AS u
ON p.user_id = u.user_id;
이 구성은:
- CSV 파일에서 읽는 시간 테이블 소스
users를 정의하고, - 예: 작업 이름 같은 속성을 설정하고,
- savepoint 경로를 설정하고,
- 지정된 savepoint 경로에서 savepoint 를 로드하는 SQL 작업을 제출합니다.
주의: 대화형 모드와 비교해 SQL Client 는 오류가 있을 때 실행을 중지하고 종료합니다.
SQL 파일을 Application Cluster 에 배포
SQL Client 는 config.yaml 또는 시작 옵션에서 배포 대상을 지정하면 -f 옵션으로 SQL 스크립트 파일을 Application Cluster 에 배포하는 것도 지원합니다. 다음은 Application Cluster 에 스크립트 파일을 배포하는 예입니다:
./bin/sql-client.sh -f oss://path/to/script.sql \
-Dexecution.target=kubernetes-application \
-Dkubernetes.cluster-id=${CLUSTER_ID} \
-Dkubernetes.container.image.ref=${FLINK_IMAGE_NAME}'
실행 후 SQL Client 는 터미널에 클러스터 id 를 인쇄합니다. 스크립트는 Flink 가 지원하는 모든 문을 포함할 수 있습니다. 그러나 Application 클러스터는 하나의 작업만 지원합니다. 제한 사항은 Application Mode 를 참고하세요.
주의: 스크립트를 클러스터에 배포할 때 SQL Client 는
--jars시작 옵션으로만 실행을 지원하며--init같은 다른 옵션은 지원되지 않습니다.
SQL 문 집합 실행
SQL Client 는 각 INSERT INTO 문을 단일 Flink 작업으로 실행합니다. 그러나 이는 때때로 파이프라인의 일부를 재사용할 수 있으므로 최적이 아닙니다. SQL Client 는 SQL 문 집합을 실행하기 위해 STATEMENT SET 문법을 지원합니다. 이는 Table API 의 StatementSet 과 동등한 기능입니다. STATEMENT SET 문법은 하나 이상의 INSERT INTO 문을 묶습니다. STATEMENT SET 블록의 모든 문은 전체적으로 최적화되어 단일 Flink 작업으로 실행됩니다. 결합된 최적화와 실행은 공통 중간 결과의 재사용을 허용하므로 여러 쿼리 실행의 효율성을 크게 향상시킬 수 있습니다.
문법
EXECUTE STATEMENT SET
BEGIN
-- one or more INSERT INTO statements
{ INSERT INTO|OVERWRITE <select_statement>; }+
END;
주의:
STATEMENT SET에 포함된 문은 세미콜론(;)으로 구분되어야 합니다. 이전 문법BEGIN STATEMENT SET; ... END;는 더 이상 사용되지 않으며 향후 버전에서 제거될 수 있습니다.
SQL CLI:
Flink SQL> CREATE TABLE pageviews (
> user_id BIGINT,
> page_id BIGINT,
> viewtime TIMESTAMP,
> proctime AS PROCTIME()
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'pageviews',
> 'properties.bootstrap.servers' = '...',
> 'format' = 'avro'
> );
[INFO] Execute statement succeeded.
Flink SQL> CREATE TABLE pageview (
> page_id BIGINT,
> cnt BIGINT
> ) WITH (
> 'connector' = 'jdbc',
> 'url' = 'jdbc:mysql://localhost:3306/mydatabase',
> 'table-name' = 'pageview'
> );
[INFO] Execute statement succeeded.
Flink SQL> CREATE TABLE uniqueview (
> page_id BIGINT,
> cnt BIGINT
> ) WITH (
> 'connector' = 'jdbc',
> 'url' = 'jdbc:mysql://localhost:3306/mydatabase',
> 'table-name' = 'uniqueview'
> );
[INFO] Execute statement succeeded.
Flink SQL> EXECUTE STATEMENT SET
> BEGIN
>
> INSERT INTO pageview
> SELECT page_id, count(1)
> FROM pageviews
> GROUP BY page_id;
>
> INSERT INTO uniqueview
> SELECT page_id, count(distinct user_id)
> FROM pageviews
> GROUP BY page_id;
>
> END;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 6b1af540c0c0bb3fcfcad50ac037c862
SQL 파일:
CREATE TABLE pageviews (
user_id BIGINT,
page_id BIGINT,
viewtime TIMESTAMP,
proctime AS PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'pageviews',
'properties.bootstrap.servers' = '...',
'format' = 'avro'
);
CREATE TABLE pageview (
page_id BIGINT,
cnt BIGINT
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/mydatabase',
'table-name' = 'pageview'
);
CREATE TABLE uniqueview (
page_id BIGINT,
cnt BIGINT
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/mydatabase',
'table-name' = 'uniqueview'
);
EXECUTE STATEMENT SET
BEGIN
INSERT INTO pageview
SELECT page_id, count(1)
FROM pageviews
GROUP BY page_id;
INSERT INTO uniqueview
SELECT page_id, count(distinct user_id)
FROM pageviews
GROUP BY page_id;
END;
DML 문을 동기/비동기로 실행
기본적으로 SQL Client 는 DML 문을 비동기로 실행합니다. 즉 SQL Client 는 DML 문에 대한 작업을 Flink 클러스터에 제출하고 작업이 끝날 때까지 기다리지 않습니다. 따라서 SQL Client 는 동시에 여러 작업을 제출할 수 있습니다. 이는 일반적으로 장기 실행되는 스트리밍 작업에 유용합니다.
SQL Client 는 문이 클러스터에 성공적으로 제출되도록 보장합니다. 문이 제출되면 CLI 는 Flink 작업에 대한 정보를 보여줍니다.
Flink SQL> INSERT INTO MyTableSink SELECT * FROM MyTableSource;
[INFO] Table update statement has been successfully submitted to the cluster:
Cluster ID: StandaloneClusterId
Job ID: 6f922fe5cba87406ff23ae4a7bb79044
주의: SQL Client 는 제출 후 실행 중인 Flink 작업의 상태를 추적하지 않습니다. 제출 후 CLI 프로세스를 종료해도 분리형 쿼리에는 영향을 주지 않습니다. Flink 의
restart strategy가 내결함성을 처리합니다. 분리형 쿼리 상태 모니터링 또는 분리형 쿼리 중지 에는 작업 문을 사용하세요.
그러나 배치 사용자의 경우 다음 DML 문이 이전 DML 문이 끝날 때까지 기다려야 하는 것이 더 일반적입니다. DML 문을 동기식으로 실행하려면 SQL Client 에서 table.dml-sync 옵션을 true 로 설정할 수 있습니다.
Flink SQL> SET 'table.dml-sync' = 'true';
[INFO] Session property has been set.
Flink SQL> INSERT INTO MyTableSink SELECT * FROM MyTableSource;
[INFO] Submitting SQL update statement to the cluster...
[INFO] Execute statement in sync mode. Please wait for the execution finish...
[INFO] Complete execution of the SQL update statement.
주의: 작업을 종료하려면
CTRL-C를 입력해 실행을 취소하세요.
savepoint 에서 SQL 작업 시작
Flink 는 지정된 savepoint 로 작업을 시작하는 것을 지원합니다. SQL Client 에서 SET 명령을 사용해 savepoint 의 경로를 지정할 수 있습니다.
Flink SQL> SET 'execution.state-recovery.path' = '/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab';
[INFO] Session property has been set.
-- all the following DML statements will be restroed from the specified savepoint path
Flink SQL> INSERT INTO ...
savepoint 경로가 지정되면 Flink 는 이후 모든 DML 문을 실행할 때 savepoint 에서 상태를 복원하려 시도합니다.
지정된 savepoint 경로가 이후 모든 DML 문에 영향을 주기 때문에 RESET 명령으로 이 구성 옵션을 재설정할 수 있습니다. 즉 savepoint 에서 복원을 비활성화합니다.
Flink SQL> RESET 'execution.state-recovery.path';
[INFO] Session property has been reset.
savepoint 생성 및 관리에 대한 자세한 내용은 Job Lifecycle Management 를 참고하세요.
사용자 정의 작업 이름 정의
SQL Client 는 SET 명령을 통해 쿼리 및 DML 문의 작업 이름 정의를 지원합니다.
Flink SQL> SET 'pipeline.name' = 'kafka-to-hive';
[INFO] Session property has been set.
-- all the following DML statements will use the specified job name.
Flink SQL> INSERT INTO ...
지정된 작업 이름이 이후 모든 쿼리와 DML 문에 영향을 주기 때문에 RESET 명령으로 이 구성을 재설정할 수도 있습니다. 즉 기본 작업 이름을 사용합니다.
Flink SQL> RESET 'pipeline.name';
[INFO] Session property has been reset.
pipeline.name 옵션이 지정되지 않으면 SQL Client 는 제출된 작업에 기본 이름을 생성합니다. 예: INSERT INTO 문의 경우 insert-into_<sink_table_name>.
작업 상태 모니터링
SQL Client 는 SHOW JOBS 문을 통해 클러스터의 작업 상태 목록을 지원합니다.
Flink SQL> SHOW JOBS;
+----------------------------------+---------------+----------+-------------------------+
| job id | job name | status | start time |
+----------------------------------+---------------+----------+-------------------------+
| 228d70913eab60dda85c5e7f78b5782c | kafka-to-hive | RUNNING | 2023-02-11T05:03:51.523 |
+----------------------------------+---------------+----------+-------------------------+
작업 종료
SQL Client 는 STOP JOB 문을 통해 savepoint 유무에 관계없이 작업 중지를 지원합니다.
Flink SQL> STOP JOB '228d70913eab60dda85c5e7f78b5782c' WITH SAVEPOINT;
+-----------------------------------------+
| savepoint path |
+-----------------------------------------+
| file:/tmp/savepoint-3addd4-0b224d9311e6 |
+-----------------------------------------+
savepoint 경로는 클러스터 구성 또는 세션 구성(후자가 우선함)에서 execution.checkpointing.savepoint-dir 로 지정할 수 있습니다.
작업 중지에 대한 자세한 내용은 Job Statements 를 참고하세요.
SQL 문법 하이라이트
SQL Client 는 여러 색상 구성표로 SQL 문법을 하이라이트할 수 있습니다. sql-client.display.color-schema 로 색상 구성표를 설정할 수 있습니다. 사용 가능한 색상 구성표: chester, dracula, solarized, vs2010, obsidian, geshi, dark, light, default (하이라이트 없음). 이름이 잘못된 경우 폴백은 default 입니다.
| 색상 스키마 \ 스타일 | Keyword | Default | Comment | Hint | Quoted | SQL Identifier |
|---|---|---|---|---|---|---|
Default |
Default | Default | Default | Default | Default | Default |
Chester |
Bold blue | White | Italic green | Bold green | Red | Cyan |
Dark |
Bold blue | White | Italic bright | Bold bright | Green | Cyan |
Dracula |
Bold magenta | White | Italic cyan | Bold cyan | Green | Red |
Geshi |
Bold #993333 | White | Italic #808080 | Bold #808080 | #66CC66 | #000099 |
Light |
Bold red | Black | Italic bright | Bold bright | Green | Cyan |
Obsidian |
Bold green | White | Italic bright | Bold bright | Red | Magenta |
VS2010 |
Bold blue | White | Italic green | Bold green | Red | Magenta |
Solarized |
Bold yellow | Blue | Italic bright | Bold bright | Green | Red |