范围中的非重复随机值

在Google Cloud Dataflow中，如果你想要从范围中生成非重复的随机值，你可以使用以下方法：

1. 使用Java的`Set`集合

Java的Set集合是一个不允许重复元素的集合。你可以使用Set来存储生成的随机数，以确保它们是唯一的。

import java.util.HashSet;
import java.util.Random;
import java.util.Set;

public class UniqueRandomNumbers {
    public static void main(String[] args) {
        int min = 1;
        int max = 100;
        int count = 10; // 你想生成的随机数的数量
        Set<Integer> uniqueNumbers = new HashSet<>();

        Random random = new Random();
        while (uniqueNumbers.size() < count) {
            int randomNumber = random.nextInt(max - min + 1) + min;
            uniqueNumbers.add(randomNumber);
        }

        System.out.println("Generated unique random numbers: " + uniqueNumbers);
    }
}

2. 使用Apache Beam的`Reshuffle`操作

在Apache Beam中，你可以使用Reshuffle操作来重新分区数据，这可以帮助你在数据处理过程中保持元素的唯一性。

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Reshuffle;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;

public class UniqueRandomNumbersPipeline {
    public static void main(String[] args) {
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);

        PCollection<Integer> numbers = pipeline.apply(Create.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

        PCollection<KV<Integer, Integer>> uniqueNumbers = numbers.apply(
            ParDo.of(new DoFn<Integer, KV<Integer, Integer>>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                    int number = c.element();
                    c.output(KV.of(number, number));
                }
            }))
            .apply(Reshuffle.<Integer, Integer>viaRandomKey())
            .apply(ParDo.of(new DoFn<KV<Integer, Integer>, Integer>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                    c.output(c.element().getKey());
                }
            }));

        uniqueNumbers.apply(BigQueryIO.writeTableRows()
            .to("your-project-id:your_dataset.unique_random_numbers")
            .withSchema(schema)
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

        pipeline.run().waitUntilFinish();
    }
}

在这个例子中，我们首先创建了一个包含数字的PCollection，然后使用ParDo将每个数字转换为一个键值对，其中键和值都是数字本身。接着，我们使用Reshuffle操作重新分区数据，以确保每个键值对的键是唯一的。最后，我们使用BigQueryIO.writeTableRows()将结果写入BigQuery表。