# 类型转换函数
#### 语法格式
```
CAST(value AS type)
```
#### 语法说明
类型强制转换。
#### 注意事项
- 若输入为NULL，则返回NULL。
- cast函数不支持将字符串转换为json对象类型。
 
#### 示例一：将amount值转换成整型
将amount值转换成整型。
```
insert into temp select cast(amount as INT) from source_stream;
```
表1类型转换函数示例 
| 示例                    | 说明                                         | 示例                                                                                                                                                                                                                                                                                                                                                          |
|:---|:---|:---|
| cast(v1 as string)    | 将v1转换为字符串类型，v1可以是数值类型，TIMESTAMP/DATE/TIME。 | 表T1: ``` | content (INT)           | | -------------           | | 5                       | ``` 语句： ``` SELECT cast(content as varchar) FROM T1; ``` 结果： ``` "5" ```                            |
| cast (v1 as int)      | 将v1转换为int, v1可以是数值类型或字符类。                  | 表T1: ``` | content  (STRING)           | | -------------               | | "5"                         | ``` 语句： ``` SELECT cast(content as int) FROM T1; ``` 结果： ``` 5 ```                           |
| cast(v1 as timestamp) | 将v1转换为timestamp类型，v1可以是字符串或DATE/TIME。      | 表T1: ``` | content  (STRING)          | | -------------              | | "2018-01-01 00:00:01"     | ``` 语句： ``` SELECT cast(content as timestamp) FROM T1; ``` 结果： ``` 1514736001000 ``` |
| cast(v1 as date)      | 将v1转换为date类型， v1可以是字符串或者TIMESTAMP。         | 表T1: ``` | content  (TIMESTAMP)     | | -------------            | | 1514736001000            | ``` 语句： ``` SELECT cast(content as date) FROM T1; ``` 结果： ``` "2018-01-01" ```                   |
   
![](https://support.huaweicloud.com/sqlref-flink-dli/public_sys-resources/note_3.0-zh-cn.png)
Flink作业不支持使用CAST将"BIGINT"转换为"TIMESTAMP"，可以使用to_timestamp进行转换。
#### 示例二
1. 参考[Kafka源表](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0386.html)和[Print结果表](https://support.huaweicloud.com/sqlref-flink-dli/dli_08_0399.html)创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。
   注意：创建作业时，在作业编辑界面的"运行参数"处，"Flink版本"选择"1.12"，勾选"保存作业日志"并设置保存作业日志的OBS桶，方便后续查看作业日志。**如下脚本中的加粗参数请根据实际环境修改** 。
   ```
   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. 连接Kafka集群，向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. 查看输出结果:
   - 方法一：
     1. 登录DLI管理控制台，选择"作业管理 \> Flink作业"。
     
     2. 在对应Flink作业所在行的"操作"列，选择"更多 \> FlinkUI"。
     
     3. 在FlinkUI界面，选择"Task Managers"，单击对应的任务名称，选择"Stdout"查看作业运行日志。
      
   
   - 方法二：若在提交运行作业前"运行参数"选择了"保存作业日志"，可以通过如下操作查看。
     1. 登录DLI管理控制台，选择"作业管理 \> Flink作业"。
     
     2. 单击对应的Flink作业名称，选择"运行日志"，单击"OBS桶"，根据作业运行的日期，找到对应日志的文件夹。
     
     3. 进入对应日期的文件夹后，找到名字中包含"taskmanager"的文件夹进入，下载获取taskmanager.out文件查看结果日志。
      
   
   
   查询结果参考如下：
   ```
   +I(1,1,2022-04-02T15:00,2022-04-02)
   ```