Type Conversion Functions

Syntax

CAST(value AS type)

Description

This function is used to forcibly convert types.

Precautions

• If the input is NULL, NULL is returned.
• The cast function does not support converting a string to the JSON format.

Example 1: Convert the amount value to an integer.

The following example converts the amount value to an integer.

insert into temp select cast(amount as INT) from source_stream;

| content (INT)           |
| -------------           |
| 5                       |

SELECT
  cast(content as varchar)
FROM
  T1;

"5"

| content  (STRING)           |
| -------------               |
| "5"                         |

SELECT
  cast(content as int)
FROM
  T1;

5

| content  (STRING)          |
| -------------              |
| "2018-01-01 00:00:01"     |

SELECT
  cast(content as timestamp)
FROM
  T1;

1514736001000

| content  (TIMESTAMP)     |
| -------------            |
| 1514736001000            |

SELECT
  cast(content as date)
FROM
  T1;

"2018-01-01"

Flink jobs do not support the conversion of bigint to timestamp using CAST. You can convert it using to_timestamp.

Example 2:

1. Create a Flink opensource SQL job by referring to Kafka Source Table and Print Result Table, enter the following job running script, and submit the job.
   Note: When creating a job, set Flink Version to 1.12 in the Running Parameters area on the job editing page, select Save Job Log, and set the OBS bucket for saving job logs to facilitate subsequent job log viewing. Change the values of the parameters in bold in the following script according to the actual situation.

   CREATE TABLE kafkaSource (
     cast_int_to_string int, 
     cast_String_to_int string,
     case_string_to_timestamp string, 
     case_timestamp_to_date timestamp
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'KafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     "format" = "json"
   );
   
   CREATE TABLE printSink (
     cast_int_to_string string, 
     cast_String_to_int int, 
     case_string_to_timestamp timestamp, 
     case_timestamp_to_date date
   ) WITH (
     'connector' = 'print'
   );
   
   insert into printSink select 
     cast(cast_int_to_string as string),
     cast(cast_String_to_int as int),
     cast(case_string_to_timestamp as timestamp),  
     cast(case_timestamp_to_date as date)
   from kafkaSource;

2. Connect to the Kafka cluster and send the following test data to the Kafka topic:

   {"cast_int_to_string":"1", "cast_String_to_int": "1", "case_string_to_timestamp": "2022-04-02 15:00:00", "case_timestamp_to_date": "2022-04-02 15:00:00"}

3. View output.
   • Method 1:
     1. Log in to the DLI management console and choose Job Management > Flink Streaming Jobs.
     2. Locate the row that contains the target Flink job, and choose More & > FlinkUI in the Operation column.
     3. On the Flink UI, choose Task Managers, click the task name, and select Stdout to view the job run logs.
   • Method 2: If you select Save Job Log for Running Parameters before submitting the job, perform the following operations:
     1. Log in to the DLI management console and choose Job Management > Flink Streaming Jobs.
     2. Click the name of the corresponding Flink job, choose Run Log, click OBS Bucket, and locate the folder of the corresponding log based on the job running date.
     3. Go to the folder of the corresponding date, find the folder whose name contains taskmanager, download the taskmanager.out file, and view the result log.
   The query result is as follows:

   +I(1,1,2022-04-02T15:00,2022-04-02)