自定义结果表

   
   


    
    
public abstract class CustomSinkBase implements Serializable{

    
    
    protected Map<String,String> userParamsMap; // 您在sql with语句中定义的键值对，但所有的键均为小写

    
    
    protected Set<String> primaryKeys; // 您定义的主键字段名

    
    
    protected List<String> headerFields;// 标记为header的字段列表

    
    
    protected RowTypeInfo rowTypeInfo;// 字段类型和名称

    
    


    
    
    /**

    
    
     * 初始化方法

    
    
     * @param taskNumber 当前节点是第几个并发

    
    
     * @param numTasks   sink节点的并发数

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract void open(int taskNumber, int numTasks) throws IOException;

    
    


    
    
    /**

    
    
     * close方法，释放资源

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract void close() throws IOException;

    
    


    
    
    /**

    
    
     * 处理插入单行数据, Row中按到ddl定义的顺序

    
    
     *

    
    
     * @param row

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract void writeAddRecord(Row row) throws IOException;

    
    


    
    
    /**

    
    
     * 处理删除单行数据，delete产生的原因可以参考：https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming.html

    
    
     * @param row

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract void writeDeleteRecord(Row row) throws IOException;

    
    


    
    
    /**

    
    
     * 如果有攒批进行批量插入，需要在该方法中实现清空该节点操作

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract void sync() throws IOException;

    
    


    
    
    /**

    
    
     * 返回用来标识sink的名字

    
    
     * @throws IOException

    
    
     */

    
    
    public abstract String getName();

    
    


    
    
}
   
   

   
   


    
    
public class UDPrintSink extends CustomSinkBase {

    
    
    private static Logger LOG = LoggerFactory.getLogger(UDPrintSink.class);

    
    


    
    
    public void open(int taskNumber, int numTasks) throws IOException {

    
    
        LOG.info(String.format("Open Method Called: taskNumber %d numTasks %d", taskNumber, numTasks));

    
    
        String[] filedNames = rowTypeInfo.getFieldNames();

    
    
        TypeInformation[] typeInformations = rowTypeInfo.getFieldTypes();

    
    
        LOG.info(String.format("Open Method Called: filedNames %d typeInformations %d", filedNames.length, typeInformations.length));

    
    
    }

    
    


    
    
    public void close() throws IOException {

    
    
        LOG.info(String.format("Close Method Called"));

    
    
    }

    
    


    
    
    public void writeAddRecord(Row row) throws IOException {

    
    
        LOG.info("Write: " + row.toString());

    
    
    }

    
    


    
    
    public void writeDeleteRecord(Row row) throws IOException {

    
    
        LOG.info("Delete: " + row.toString());

    
    
    }

    
    


    
    
    public void sync() throws IOException {

    
    
        //没有做攒批写入，空置该方法

    
    
    }

    
    


    
    
    public String getName() {

    
    
        return "UDPrintSink";

    
    
    }

    
    
}
   
   

   
   


    
    
create table customPrint (

    
    
    a int,

    
    
    b BIGINT,

    
    
    c VARCHAR

    
    
) with (

    
    
    type = 'custom',

    
    
    class = 'com.alibaba.blink.connector.custom.demo.UDPrintSink'

    
    
    [,...]

    
    
);
   
   

   
   


    
    
--源 DDL

    
    
create table suorce_name  (

    
    
    a int,

    
    
    b BIGINT

    
    
) with (

    
    
    type = 'XXXX'

    
    
    ......

    
    
);

    
    
--自定义sink ddl

    
    
create table customPrint (

    
    
    a int,

    
    
    b BIGINT,

    
    
    PRIMARY KEY(a)

    
    
) with (

    
    
    type = 'custom',

    
    
    class = 'UDPrintSink' --类名

    
    
);
   
   

   
   


    
    
INSERT INTO customPrint

    
    
SELECT 

    
    
* 

    
    
FROM suorce_name