from __future__ import print_function

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, col, coalesce

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: c19.py <input> <output>", file=sys.stderr)
        exit(-1)
    spark = SparkSession.builder.appName('Select c19').getOrCreate()
    raw = spark.read.option('mergeSchema','true').load(sys.argv[1])
    df = raw.filter(col('date') < '1900')

    opens = df.filter(col('open') == 'true')\
              .select('series', 'date', lit(1).alias('inopen')).distinct()

    df.join(opens, ['series', 'date'], 'left_outer')\
      .filter((col('open') == 'true') | col('inopen').isNull())\
      .drop('inopen')\
      .dropDuplicates(['id'])\
      .write.save(sys.argv[2])

    spark.stop()

from __future__ import print_function

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, struct, max

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: c19.py <input> <output>", file=sys.stderr)
        exit(-1)
    spark = SparkSession.builder.appName('Select c19').getOrCreate()
    raw = spark.read.option('mergeSchema','true').load(sys.argv[1])
    df = raw.filter(col('date') < '1900')

    spark.conf.set('spark.sql.shuffle.partitions', df.rdd.getNumPartitions() * 2)

    issues = df.groupBy('series', 'date')\
               .agg(max(struct('open', 'corpus'))['corpus'].alias('corpus'))
    
    df.join(issues, ['series', 'date', 'corpus'], 'inner')\
      .write.save(sys.argv[2])

    spark.stop()

Choose a single corpus for a given series,date pair.
Choose a single corpus for a given series,date pair.

  from __future__ import print_function
  
  import sys
  
  from pyspark.sql import SparkSession
- from pyspark.sql.functions import lit, col, coalesce
+ from pyspark.sql.functions import col, struct, max
  
  if __name__ == "__main__":
      if len(sys.argv) != 3:
          print("Usage: c19.py <input> <output>", file=sys.stderr)
          exit(-1)
      spark = SparkSession.builder.appName('Select c19').getOrCreate()
      raw = spark.read.option('mergeSchema','true').load(sys.argv[1])
      df = raw.filter(col('date') < '1900')
  
+     spark.conf.set('spark.sql.shuffle.partitions', df.rdd.getNumPartitions() * 2)
-     opens = df.filter(col('open') == 'true')\
-               .select('series', 'date', lit(1).alias('inopen')).distinct()
  
+     issues = df.groupBy('series', 'date')\
+                .agg(max(struct('open', 'corpus'))['corpus'].alias('corpus'))
+     
-     df.join(opens, ['series', 'date'], 'left_outer')\
+     df.join(issues, ['series', 'date', 'corpus'], 'inner')\
-       .filter((col('open') == 'true') | col('inopen').isNull())\
-       .drop('inopen')\
-       .dropDuplicates(['id'])\
        .write.save(sys.argv[2])
  
      spark.stop()
  
Choose a single corpus for a given series,date pair.
## Code Before:
from __future__ import print_function

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, col, coalesce

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: c19.py <input> <output>", file=sys.stderr)
        exit(-1)
    spark = SparkSession.builder.appName('Select c19').getOrCreate()
    raw = spark.read.option('mergeSchema','true').load(sys.argv[1])
    df = raw.filter(col('date') < '1900')

    opens = df.filter(col('open') == 'true')\
              .select('series', 'date', lit(1).alias('inopen')).distinct()

    df.join(opens, ['series', 'date'], 'left_outer')\
      .filter((col('open') == 'true') | col('inopen').isNull())\
      .drop('inopen')\
      .dropDuplicates(['id'])\
      .write.save(sys.argv[2])

    spark.stop()

## Instruction:
Choose a single corpus for a given series,date pair.
## Code After:
from __future__ import print_function

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, struct, max

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: c19.py <input> <output>", file=sys.stderr)
        exit(-1)
    spark = SparkSession.builder.appName('Select c19').getOrCreate()
    raw = spark.read.option('mergeSchema','true').load(sys.argv[1])
    df = raw.filter(col('date') < '1900')

    spark.conf.set('spark.sql.shuffle.partitions', df.rdd.getNumPartitions() * 2)

    issues = df.groupBy('series', 'date')\
               .agg(max(struct('open', 'corpus'))['corpus'].alias('corpus'))
    
    df.join(issues, ['series', 'date', 'corpus'], 'inner')\
      .write.save(sys.argv[2])

    spark.stop()
