<jupyter_start><jupyter_text>Connect to PySpark master<jupyter_code>builder = (SparkSession
           .builder
           .appName('PySpark Example')
           .master('spark://carbon.uedalab.local:7077')
           .config("spark.jars.packages", "org.diana-hep:spark-root_2.11:0.1.15")
           .config("spark.cores.max", 1)
           )
spark = builder.getOrCreate()<jupyter_output><empty_output><jupyter_text>Read root files<jupyter_code>globbed = iglob("/helium/analysis/SPring-8/2014B/06_Ar_15atm_258eV/ResortLi/ResortLess000[123].root")
loaded = (spark.read.format("org.dianahep.sparkroot").load(fn) for fn in globbed)
df = reduce(DataFrame.union, loaded)
df.printSchema()<jupyter_output>root
 |-- IonNum: integer (nullable = true)
 |-- IonX0: double (nullable = true)
 |-- IonY0: double (nullable = true)
 |-- IonT0: double (nullable = true)
 |-- IonFlag0: integer (nullable = true)
 |-- IonX1: double (nullable = true)
 |-- IonY1: double (nullable = true)
 |-- IonT1: double (nullable = true)
 |-- IonFlag1: integer (nullable = true)
 |-- IonX2: double (nullable = true)
 |-- IonY2: double (nullable = true)
 |-- IonT2: double (nullable = true)
 |-- IonFlag2: integer (nullable = true)
 |-- IonX3: double (nullable = true)
 |-- IonY3: double (nullable = true)
 |-- IonT3: double (nullable = true)
 |-- IonFlag3: integer (nullable = true)
 |-- ElecNum: integer (nullable = true)
 |-- ElecX0: double (nullable = true)
 |-- ElecY0: double (nullable = true)
 |-- ElecT0: double (nullable = true)
 |-- ElecFlag0: integer (nullable = true)
 |-- ElecX1: double (nullable = true)
 |-- ElecY1: double (nullable = true)
 |-- ElecT1: double (nullable = true)
 |-- ElecFlag1: integer (nullable = tr[...]<jupyter_text>Restruct<jupyter_code>@udf(SpkHits)
def combine_hits(xarr, yarr, tarr, flagarr, nhits):
    zipped = ({'x': x,
               'y': y,
               't': t,
               'flag': f
               } for x, y, t, f in zip(xarr, yarr, tarr, flagarr))
    return list(islice(zipped, nhits))


imhits = sum(1 for c in df.columns if c.startswith('IonT'))
emhits = sum(1 for c in df.columns if c.startswith('ElecT'))
restructured = (
    df
        .withColumn('itarr', array(*['IonT{}'.format(i) for i in range(imhits)]))
        .withColumn('ixarr', array(*['IonX{}'.format(i) for i in range(imhits)]))
        .withColumn('iyarr', array(*['IonY{}'.format(i) for i in range(imhits)]))
        .withColumn('iflagarr', array(*['IonFlag{}'.format(i) for i in range(imhits)]))
        .withColumnRenamed('IonNum', 'inhits')
        .withColumn('etarr', array(*['ElecT{}'.format(i) for i in range(emhits)]))
        .withColumn('exarr', array(*['ElecX{}'.format(i) for i in range(emhits)]))
        .withColumn('eyarr', array(*['ElecY{}'.format(i) for i in range(emhits)]))
        .withColumn('eflagarr', array(*['ElecFlag{}'.format(i) for i in range(emhits)]))
        .withColumnRenamed('ElecNum', 'enhits')
        .select(combine_hits('itarr', 'ixarr', 'iyarr', 'iflagarr', 'inhits').alias('ihits'),
                combine_hits('etarr', 'exarr', 'eyarr', 'eflagarr', 'enhits').alias('ehits'))
)
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18629006664550937, 'fraction_numerical': 0.017772135829895272, 'mean_word_length': 3.548340548340548, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '539084', 'license': 'MIT', 'n_tokens_mistral': 1213, 'n_tokens_neox': 1180, 'n_words': 241, 'path': 'Examples/Read SP8 root files.ipynb', 'repo_name': 'DaehyunYou/sp8-delayline', 'size': '3151'}