/flinksql-connector-oss

Flink SQL connector for OSS database

Primary LanguageJavaMIT LicenseMIT

Flink OSS Connector

Flink SQL connector for OSS database, this project Powered by OSS Java SDK. This is a connector that implements the most basic functions. Better and richer functions can be added on the basis of this connector.

Connector Options

Option Required Default Type Description
endpoint required none String Endpoint indicates the OSS external service access domain name.
access-key-id required none String AccessKey refers to the accessKeyId and accessKeySecret used in access authentication.
access-key-secret required none String
bucket-name required none String Storage space is a container used by users to store objects.
object-name required none String Objects are the basic units of OSS data storage, also known as OSS files.
format required none String Flink provides a set of table formats that can be used with table connectors.

How to use

Read a csv file on OSS

CREATE TABLE sourceTable (
    name STRING,
    age INT)
WITH (
  'connector' = 'oss',
  'endpoint' = '<your oss endpoint>',
  'access-key-id' = '<your oss accessKeyId>',
  'access-key-secret' = '<your oss accessKeySecret>',
  'bucket-name' = '<your oss bucketName>',
  'object-name' = '<your oss objectName>>',
  'format' = 'csv'
);

CREATE TABLE sinkTable (
    name VARCHAR,
    age INT
) WITH (
    'connector' = 'print'
);

INSERT INTO sinkTable(name,age) SELECT name,age FROM sourceTable limit 10;

Write a CSV file to OSS

CREATE TABLE sourceTable (
    f_random INT,
    f_random_str STRING
    ) WITH (
    'connector' = 'datagen',
    'rows-per-second'='1',
    'number-of-rows'='10',
    'fields.f_random.max'='100',
    'fields.f_random.min'='0',
    'fields.f_random_str.length'='3'
);


CREATE TABLE sinkTable (
    name STRING,
    age INT)
WITH (
  'connector' = 'oss',
  'endpoint' = '<your oss endpoint>',
  'access-key-id' = '<your oss accessKeyId>',
  'access-key-secret' = '<your oss accessKeySecret>',
  'bucket-name' = '<your oss bucketName>',
  'object-name' = '<your oss objectName>>',
  'format' = 'csv',
  'csv.field-delimiter' = ','
);

INSERT INTO sinkTable(name,age) SELECT f_random_str,f_random FROM sourceTable;