new_contents: import argparse
  import json, os, sys
  from math import ceil, log, inf
  from pyspark.sql import SparkSession, Row
  from pyspark.sql.functions import (col, desc, explode, size, udf, struct, length,
                                     collect_list, collect_set, sort_array, when,
                                     expr, map_from_entries, flatten, xxhash64, lit,
                                     array, arrays_zip)
  import pyspark.sql.functions as f
  
  def main(config):
      spark = SparkSession.builder.appName('Sample Clusters').getOrCreate()
  
      raw = spark.read.load(config.inputPath)
  
      samples = raw.filter(config.filter).groupBy('cluster').agg(
          f.min(struct('date', 'uid'))['uid'].alias('uid')).sample(fraction=0.01
                                                                   ).limit(config.samples)
      raw.join(samples, ['cluster', 'uid'], 'left_semi'
              ).repartition(1).sort('date', 'begin'
              ).coalesce(1).write.csv(config.outputPath,
                                      escape='"',
                                      header=True)
      
      spark.stop()
  
  if __name__ == '__main__':
      parser = argparse.ArgumentParser(description='Sample Clusters',
                                       formatter_class=argparse.ArgumentDefaultsHelpFormatter)
      parser.add_argument('-f', '--filter', type=str, default='size >= 10 AND pboiler < 0.2',
                          help='SQL query for reprints; default=size >= 10 AND pboiler < 0.2')
      parser.add_argument('-s', '--samples', type=int, default=1000,
                          help='Number of samples; default=1000')
      parser.add_argument('inputPath', metavar='<path>', help='input path')
      parser.add_argument('outputPath', metavar='<path>', help='output path')
  
      config = parser.parse_args()
      print(config)
      main(config)
  
subject: Add sample-clusters.py to sample at most one text from each cluster.