FileSystem源表
语法格式
1 2 3 4 5 6 7 8 9 10 11 |
CREATE TABLE sink_table ( name string, num INT, p_day string, p_hour string ) partitioned by (p_day, p_hour) WITH ( 'connector' = 'filesystem', 'path' = 'obs://*** ', 'format' = 'parquet', 'source.monitor-interval'='' ); |
参数说明
- 目录监控
默认情况下,文件系统连接器是有界的,也就是只会扫描配置路径一遍后就会停止。
如果需要,可以通过设置 source.monitor-interval 属性来开启目录监控,以便在新文件出现时继续扫描。
键
默认值
类型
描述
source.monitor-interval
无
Duration
设置新文件的监控时间间隔,并且必须设置 > 0 的值。
每个文件都有其路径唯一标识,一旦发现新文件,就会处理一次。
已处理的文件在 source 的整个生命周期内存储在 state 中,因此,source 的 state 在 checkpoint 和 savepoint 时进行保存。
更短的时间间隔意味着文件被更快地发现,但也意味着更频繁地遍历文件系统/对象存储。
如果未设置此配置选项,则提供的路径仅被扫描一次,因此源将是有界的。
- 可用的Metadata
以下连接器 metadata 可以在表定义时作为 metadata 列进行访问。所有 metadata 都是只读的。
键
数据类型
描述
file.path
STRING NOT NULL
输入文件的完整路径。
file.name
STRING NOT NULL
文件名,即距离文件根路径最远的元素。
file.size
STRING NOT NULL
文件的字节数。
file.modification-time
TIMESTAMP_LTZ(3) NOT NULL
文件的修改时间。
示例
从obs表作为数据源读取数据,输出到print connector。
CREATE TABLE obs_source( name string, num INT, `file.path` STRING NOT NULL METADATA ) WITH ( 'connector' = 'filesystem', 'path' = 'obs://demo/sink_parquent_obs', 'format' = 'parquet', 'source.monitor-interval'='1 h' ); CREATE TABLE print ( name string, num INT, path STRING ) WITH ( 'connector' = 'print' ); insert into print select * from obs_source;
+I[0e72e, 841255524, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0] +I[53524, -2032270969, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0] +I[77225, 245599258, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0] +I[fc202, -545621464, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0] +I[07e9d, 1511139764, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0] +I[4e48b, 278014413, /spark.db/sink_parquent_obs/compacted-part-fd4d4cc8-8b18-42d5-b522-9b524500fa23-0-0]