#!/usr/bin/env python3
"""Delta Lake na HDFS: wynik zliczania słów jako tabela z historią wersji.

Kroki: zapis tabeli (wersja 0), DELETE (wersja 1), UPDATE (wersja 2), DESCRIBE HISTORY,
podróż w czasie (VERSION AS OF 0) i porównanie z bieżącą wersją.

Użycie:  python delta_demo.py
Wymaga pakietu delta-spark (pip install delta-spark). Skrypt uruchamiamy zwykłym `python`, a nie
`spark-submit`, bo `configure_spark_with_delta_pip` dopisuje do konfiguracji pakiet Delta, który
Spark pobiera z Maven Central przy pierwszym uruchomieniu (JVM startuje już z tym pakietem).
"""
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

BASE_PATH = 'hdfs://127.0.0.1:9000/user/hadoop'
INPUT_PATH = BASE_PATH + '/pan_tadeusz/small'
TABLE_PATH = BASE_PATH + '/delta/pan_tadeusz_words'
T = 'delta.`%s`' % TABLE_PATH      # tabela Delta wskazana ścieżką, bez rejestrowania w katalogu


def section(title):
    print('\n' + '=' * 78 + '\n' + title + '\n' + '=' * 78)


builder = (SparkSession.builder.appName('Delta Lake Pan Tadeusz')
           .config('spark.sql.extensions', 'io.delta.sql.DeltaSparkSessionExtension')
           .config('spark.sql.catalog.spark_catalog', 'org.apache.spark.sql.delta.catalog.DeltaCatalog'))
spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel('WARN')

# --- wersja 0: wynik liczenia zapisany jako tabela Delta
section('1. Zapis wyniku liczenia jako tabeli Delta (wersja 0)')
counts = (spark.read.text(INPUT_PATH)
          .select(F.explode(F.split(F.lower('value'), r'[^\p{L}]+')).alias('word'))
          .where("word <> ''")
          .groupBy('word').count().withColumnRenamed('count', 'cnt'))
counts.write.format('delta').mode('overwrite').save(TABLE_PATH)
spark.sql('SELECT COUNT(*) AS words, SUM(cnt) AS occurrences FROM %s' % T).show()
spark.sql('SELECT * FROM %s ORDER BY cnt DESC, word LIMIT 5' % T).show()

# --- wersja 1: DELETE
section('2. DELETE: usuwamy słowa jednoliterowe (wersja 1)')
spark.sql('SELECT word, cnt FROM %s WHERE length(word) = 1 ORDER BY cnt DESC' % T).show(5)
spark.sql('DELETE FROM %s WHERE length(word) = 1' % T).show()
spark.sql('SELECT COUNT(*) AS words, SUM(cnt) AS occurrences FROM %s' % T).show()

# --- wersja 2: UPDATE
section('3. UPDATE: dopisujemy wystąpienia z tytułu utworu, który wycięliśmy z pliku (wersja 2)')
spark.sql("SELECT word, cnt FROM %s WHERE word IN ('pan', 'tadeusz') ORDER BY word" % T).show()
spark.sql("UPDATE %s SET cnt = cnt + 1 WHERE word IN ('pan', 'tadeusz')" % T).show()
spark.sql("SELECT word, cnt FROM %s WHERE word IN ('pan', 'tadeusz') ORDER BY word" % T).show()

# --- historia
section('4. DESCRIBE HISTORY: kto, kiedy i co zmienił')
history = spark.sql('DESCRIBE HISTORY %s' % T)
history.select('version', 'timestamp', 'userName', 'operation', 'operationParameters').show(truncate=70)
history.select('version', 'operation',
               F.col('operationMetrics')['numAddedFiles'].alias('added_files'),
               F.col('operationMetrics')['numRemovedFiles'].alias('removed_files'),
               F.col('operationMetrics')['numDeletedRows'].alias('deleted_rows'),
               F.col('operationMetrics')['numUpdatedRows'].alias('updated_rows')).show()

# --- podróż w czasie
section('5. Podróż w czasie: VERSION AS OF 0 kontra bieżąca wersja')
spark.sql('SELECT COUNT(*) AS words_v0 FROM %s VERSION AS OF 0' % T).show()
spark.sql('SELECT COUNT(*) AS words_current FROM %s' % T).show()
print('Słowa, które były w wersji 0, a w bieżącej ich nie ma (10 najczęstszych):')
spark.sql('''SELECT word, cnt FROM %s VERSION AS OF 0
             WHERE word NOT IN (SELECT word FROM %s)
             ORDER BY cnt DESC, word LIMIT 10''' % (T, T)).show()
print("Słowa 'pan' i 'tadeusz' w obu wersjach:")
spark.sql('''SELECT v0.word, v0.cnt AS version_0, now.cnt AS current
             FROM %s VERSION AS OF 0 v0 JOIN %s now ON v0.word = now.word
             WHERE v0.word IN ('pan', 'tadeusz') ORDER BY v0.word''' % (T, T)).show()

spark.stop()