CREATE TABLE MyTable(
    channel varchar as sss not null,
    pv INT not null,
    xctime bigint,
    CHARACTER_LENGTH(channel) as timeLeng,
    WATERMARK FOR xctime AS withOffset(xctime,1000)
 )WITH(
    type='kafka11',
    bootstrapServers='172.16.8.107:9092',
    offsetReset='latest',
    topic='toutou'
 );

CREATE TABLE MyResult(
    channel varchar,
    pv INT
 )WITH(
    type='mysql',
    url='jdbc:mysql://172.16.8.109:3306/test?charset=utf8',
    userName='dtstack',
    password='abc123',
    tableName='pv'
 );


insert
into
    MyResult
    select
        a.channel,
        a.pv
    from
        MyTable a

