Skip to content

Latest commit

 

History

History
101 lines (73 loc) · 2.53 KB

README.md

File metadata and controls

101 lines (73 loc) · 2.53 KB
Raw

Flume PostgreSQL Sink

1 使用场景

  • 目前只适用于Flume source是kafka，后面考虑支持其他的source，支持订阅多个topic并将数据写入到topic对应的表中
  • 目前只支持kafka中message与table字段名相同的情况，后面考虑支持自定义字段映射
  • 目前只支持kafka中messages是json格式

2 快速起步

2.1 例子

#postgresql 配置
postgresql.sources=s1
postgresql.channels=c1
postgresql.sinks=k1

#s1 配置
postgresql.sources.s1.type=org.apache.flume.source.kafka.KafkaSource
postgresql.sources.s1.channels=c1
postgresql.sources.s1.batchSize=1000
postgresql.sources.s1.batchDurationMillis=2000
postgresql.sources.s1.kafka.bootstrap.servers = ip:port,ip2:port2,ip3:port3
postgresql.sources.s1.kafka.topics = topicName,topicName2
postgresql.sources.s1.kafka.consumer.group.id = flume-pg-xxxx

#c1 配置
postgresql.channels.c1.type=file
postgresql.channels.c1.checkpointDir=/data/logs/flume-pg/postgresql/file-channel/checkpoint
postgresql.channels.c1.dataDirs=/data/logs/flume-pg/postgresql/file-channel/data
postgresql.channels.c1.capacity=500000
postgresql.channels.c1.transactionCapacity=10000
postgresql.channels.c1.checkpointInterval=30000

#k1 配置
postgresql.sinks.k1.channel = c1
postgresql.sinks.k1.type = com.dh.flume.postgresql.sink.PostgreSQLSink
postgresql.sinks.k1.hostname = 192.168.xxx.xxx
postgresql.sinks.k1.port = 5432
postgresql.sinks.k1.database = databaseName
postgresql.sinks.k1.user = test
postgresql.sinks.k1.password = test
postgresql.sinks.k1.batchSize = 1000
postgresql.sinks.k1.topicToTable = [{"tableName":"tableName","writeMode":"insert","topicName":"topicName"},{"tableName":"tableName2","writeMode":"insert","topicName":"topicName2"}]

2.2 参数说明

只作sink的参数说明，source及channel参数说明请参考 http://flume.apache.org/releases/content/1.8.0/FlumeUserGuide.html

  • type

    • 描述：sink类型
      • com.dh.flume.postgresql.sink.PostgreSQLSink
    • 必选：是
    • 默认值：无

  • hostname

    • 描述：ip。
    • 必选：是
    • 默认值：无

  • port

    • 描述：端口。
    • 必选：是
    • 默认值：无

  • user

    • 描述：用户名。
    • 必选：是
    • 默认值：无

  • password

    • 描述：密码。
    • 必选：是
    • 默认值：无

  • batchSize

    • 描述：批量操作size。
    • 必选：否
    • 默认值：100

  • topicToTable

    • 描述：kafka topic与table name的映射，json array的格式。
    • 必选：是
    • 默认值：无

2.3 编译打包

gradle build -x test