如何打印流数据帧的DataSource选项(例如startingOffsets)？

在处理流数据帧时，DataSource 选项通常指的是数据源的配置信息，如 startingOffsets（起始偏移量）等。这些选项在多种流处理框架中都很常见，例如 Apache Kafka、Apache Flink 等。下面我将分别介绍在这些框架中如何打印或查看 DataSource 选项。

Apache Kafka

在 Kafka 中，startingOffsets 是消费者在开始消费时指定的起始偏移量。你可以通过 Kafka 的消费者 API 来获取这些信息。

示例代码（Java）：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        TopicPartition partition = new TopicPartition("test-topic", 0);
        consumer.assign(Collections.singletonList(partition));

        // 获取起始偏移量
        long startingOffset = consumer.beginningOffsets(Collections.singletonList(partition)).get(partition);
        System.out.println("Starting offset: " + startingOffset);

        consumer.close();
    }
}

Apache Flink

在 Flink 中，startingOffsets 可以通过 DataStream 的 setStartFromGroupOffsets() 或 setStartFromEarliest() 等方法来设置。要打印这些选项，你可以直接在代码中查看或记录这些设置。

示例代码（Java）：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class FlinkKafkaConsumerExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test-group");

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("test-topic", new SimpleStringSchema(), properties);
        kafkaConsumer.setStartFromEarliest(); // 设置起始偏移量为最早

        // 打印起始偏移量设置
        System.out.println("Starting offsets set to earliest");

        env.addSource(kafkaConsumer).print();

        env.execute("Flink Kafka Example");
    }
}