S3 Source
Documentación Oficial
Generalized Amazon S3 Source Connector for Confluent Platform
- Connector Operation Mode: RESTORE_BACKUP
Configuration Reference for Amazon S3 Sink Connector for Confluent Platform
Partitioner Parameters (Backup and Restore Connector Only)¶
partitioner.class
The partitioner to use when reading data from the store.
- Type: class
- Default: io.confluent.connect.storage.partitioner.DefaultPartitioner
- Importance: high
partition.field.name
The name of the partitioning field when FieldPartitioner is used.
- Type: list
- Default: “”
- Importance: medium
path.format
This configuration that was used to set the format of the data directories when partitioning with a TimeBasedPartitioner
- Type: string
- Default: “”
- Importance: medium
Source Safety¶
-------------------------
-- S3 Source - Safety
-------------------------
CREATE SOURCE CONNECTOR `source-safety` WITH(
"connector.class" = 'io.confluent.connect.s3.source.S3SourceConnector',
"confluent.topic.bootstrap.servers" = 'kafka:9092',
"confluent.license" = '',
"confluent.license.topic.replication.factor" = '1',
"confluent.topic.replication.factor" = '1',
"storage.class" = 'io.confluent.connect.s3.storage.S3Storage',
"format.class" = 'io.confluent.connect.s3.format.json.JsonFormat',
"topics.dir" = 'safety',
"tasks.max" = '1',
"s3.bucket.name" = '4stations',
"s3.region" = 'eu-west-2',
"aws.access.key.id" = 'YA9JokyWUb2hFUbKYEEN',
"aws.secret.access.key" = '0k2ornkQpVTqUrBbOEsEXOnBEEWgJf4AFQOU4O7Y',
"store.url" = 'https://izar.ls.fi.upm.es:30009',
"partitioner.class" = 'io.confluent.connect.storage.partitioner.TimeBasedPartitioner',
"path.format" = ''\'hour\''=HH',
"locale" = 'es-ES',
"timezone" = 'Europe/Madrid',
"key.converter" = 'org.apache.kafka.connect.storage.StringConverter',
"key.converter.schema.registry.url" = 'http://schema-registry:8081',
"key.converter.schemas.enable" = 'false',
"value.converter" = 'org.apache.kafka.connect.json.JsonConverter',
"value.converter.schema.registry.url" = 'http://schema-registry:8081',
"value.converter.schemas.enable" = 'false');
Source Traffic¶
-------------------------
-- S3 Source - Traffic
-------------------------
CREATE SOURCE CONNECTOR `source-traffic` WITH(
"connector.class" = 'io.confluent.connect.s3.source.S3SourceConnector',
"confluent.topic.bootstrap.servers" = 'kafka:9092',
"confluent.license" = '',
"confluent.license.topic.replication.factor" = '1',
"confluent.topic.replication.factor" = '1',
"storage.class" = 'io.confluent.connect.s3.storage.S3Storage',
"format.class" = 'io.confluent.connect.s3.format.avro.AvroFormat',
"topics.dir" = 'traffic',
"tasks.max" = '1',
"s3.bucket.name" = '4stations',
"s3.region" = 'eu-west-2',
"aws.access.key.id" = 'YA9JokyWUb2hFUbKYEEN',
"aws.secret.access.key" = '0k2ornkQpVTqUrBbOEsEXOnBEEWgJf4AFQOU4O7Y',
"store.url" = 'https://izar.ls.fi.upm.es:30009',
"partitioner.class" = 'io.confluent.connect.storage.partitioner.DefaultPartitioner',
"key.converter" = 'io.confluent.connect.avro.AvroConverter',
"key.converter.schema.registry.url" = 'http://schema-registry:8081',
"key.converter.schemas.enable" = 'false',
"value.converter" = 'io.confluent.connect.avro.AvroConverter',
"value.converter.schema.registry.url" = 'http://schema-registry:8081',
"value.converter.schemas.enable" = 'false');