Конвейеры данных с Apache Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 3);
props.put("batch.size", 16384);
props.put("linger.ms", 10);
props.put("buffer.memory", 33554432);
 
Producer<String, String> producer = new KafkaProducer<>(props);
 
for (int i = 0; i < 100; i++) {
    String key = "key-" + i;
    String value = "message-" + i;
    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", key, value);
    
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            System.err.println("Error sending message: " + exception.getMessage());
        } else {
            System.out.printf("Message sent to partition %d, offset %d%n", 
                              metadata.partition(), metadata.offset());
        }
    });
}
 
producer.flush();
producer.close();

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "false");
 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
 
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",
                             record.key(), record.value(), record.partition(), record.offset());
            
            // Обработка сообщения...
            
            // Ручная фиксация смещения после обработки
            Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
            offsetsToCommit.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
            consumer.commitSync(offsetsToCommit);
        }
    }
} finally {
    consumer.close();
}

// Определение схемы Avro
String userSchema = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[" +
                    "{\"name\":\"id\",\"type\":\"int\"}," +
                    "{\"name\":\"name\",\"type\":\"string\"}," +
                    "{\"name\":\"email\",\"type\":\"string\"}]}";
 
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(userSchema);
 
// Создание объекта Avro
GenericRecord user = new GenericData.Record(schema);
user.put("id", 123);
user.put("name", "John Doe");
user.put("email", "john.doe@example.com");
 
// Настройка производителя с Avro-сериализатором
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://schema-registry:8081");
 
Producer<String, GenericRecord> producer = new KafkaProducer<>(props);
ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("users-topic", "user-123", user);
producer.send(record);
producer.close();

List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
        if (buffer.size() >= 10000) {
            processInBatch(buffer);
            buffer.clear();
            consumer.commitSync();
        }
    }
}

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processImmediately(record);
        Map<TopicPartition, OffsetAndMetadata> offsetMap = Collections.singletonMap(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1)
        );
        consumer.commitSync(offsetMap);
    }
}

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> inputStream = builder.stream("input-topic");
KStream<String, String> processedStream = inputStream
    .filter((key, value) -> value.length() > 10)
    .mapValues(value -> value.toUpperCase())
    .peek((key, value) -> System.out.println("Processing: " + value));
processedStream.to("output-topic");

Properties config = new Properties();
config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 8);

config.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://schema-registry:8081");
config.put(AbstractKafkaSchemaSerDeConfig.MAX_SCHEMAS_PER_SUBJECT, 1000);

Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDBConfig.class.getName());
 
public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
        tableConfig.setBlockCacheSize(50 * 1024 * 1024L); // 50MB блочный кэш
        tableConfig.setBlockSize(32 * 1024); // 32KB блоки
        options.setTableFormatConfig(tableConfig);
        options.setMaxWriteBufferNumber(3);
        options.setWriteBufferSize(32 * 1024 * 1024); // 32MB буфер записи
    }
}

config.put(StreamsConfig.RECEIVE_BUFFER_CONFIG, 65536); // 64KB
config.put(StreamsConfig.SEND_BUFFER_CONFIG, 131072);   // 128KB

config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024L); // 10MB

TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1))

KStream<String, Transaction> transactions = builder.stream("transactions");
KStream<String, Transaction> highValueTransactions = transactions
    .filter((key, transaction) -> transaction.getAmount() > 1000);

private static final Duration PUNCTUATION_INTERVAL = Duration.ofSeconds(30);

builder.stream("input-topic")
       .selectKey((key, value) -> computeOptimalPartitionKey(value))
       .to("optimized-topic", Produced.with(Serdes.String(), valueSerde));

StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("my-store"),
    Serdes.String(),
    Serdes.Long()
).withLogConfig(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);

config.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG");

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "transactions")
  .load()
 
val query = kafkaStream
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .groupBy("key")
  .count()
  .writeStream
  .format("console")
  .outputMode("complete")
  .start()

KStream<String, StockLevel> stockStream = builder.stream("regional-stock");
KStream<String, Order> orderStream = builder.stream("customer-orders");
 
stockStream.join(
    orderStream,
    (stock, order) -> new SupplyPlanningEvent(stock, order),
    JoinWindows.of(Duration.ofMinutes(5))
).to("supply-planning");

// Пример конфигурации JDBC Source коннектора
Map<String, String> sourceProps = new HashMap<>();
sourceProps.put("connector.class", "io.confluent.connect.jdbc.JdbcSourceConnector");
sourceProps.put("connection.url", "jdbc:postgresql://localhost:5432/mydatabase");
sourceProps.put("connection.user", "postgres");
sourceProps.put("connection.password", "secret");
sourceProps.put("topic.prefix", "db_");
sourceProps.put("mode", "incrementing");
sourceProps.put("incrementing.column.name", "id");
sourceProps.put("table.whitelist", "customers,orders");

// Конфигурация HDFS Sink коннектора
Map<String, String> hdfsProps = new HashMap<>();
hdfsProps.put("connector.class", "io.confluent.connect.hdfs.HdfsSinkConnector");
hdfsProps.put("topics", "sales-transactions,customer-behaviour,promotions-usage");
hdfsProps.put("hdfs.url", "hdfs://hadoop-cluster:8020");
hdfsProps.put("flush.size", "10000");  // Записывать после каждых 10000 сообщений
hdfsProps.put("rotate.interval.ms", "3600000");  // Ротация файлов каждый час
hdfsProps.put("format.class", "io.confluent.connect.hdfs.parquet.ParquetFormat");
hdfsProps.put("partitioner.class", "io.confluent.connect.hdfs.partitioner.TimeBasedPartitioner");

// Пример интеграции Spark Streaming с Kafka
val transactionsDf = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092")
  .option("subscribe", "financial-transactions")
  .option("startingOffsets", "latest")
  .load()
 
// Десериализация JSON-данных
val parsedTransactions = transactionsDf
  .select(from_json(col("value").cast("string"), transactionSchema).as("transaction"))
  .select("transaction.*")
 
// Выявление подозрительных транзакций
val suspiciousTransactions = parsedTransactions
  .withWatermark("timestamp", "1 minute")
  .groupBy(window(col("timestamp"), "5 minutes"), col("account_id"))
  .agg(sum("amount").as("total_amount"), count("*").as("transaction_count"))
  .filter(col("total_amount") > 10000 || col("transaction_count") > 5)
 
// Запись результатов обратно в Kafka
suspiciousTransactions.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092")
  .option("topic", "suspicious-transactions-alerts")
  .option("checkpointLocation", "/checkpoints/suspicious-tx")
  .start()

props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 65536);
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-id-1");