I am trying to use Time Travel just to see how to use it, but when I run:
spark.sql("SELECT * FROM RIDES.YELLOW_TAXI VERSION AS OF 0")
It throws the error:
ParseException:
extraneous input '0' expecting {<EOF>, ';'}(line 1, pos 38)
== SQL ==
SELECT * FROM RIDES.YELLOW_TAXI AS OF 0
--------------------------------------^^^
I've tried changing the query to double quotes and placing the version number in single quotes. The really odd thing is that the query returns results when I simply run:
spark.sql("SELECT * FROM RIDES.YELLOW_TAXI AS OF")
That's just a side note though as I thought that it would fail in parsing. Data I'm using is:
https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2021-01.parquet
https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2021-02.parquet
Full code:
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.master("local[*]") \
.appName('test') \
.config("spark.jars.packages", "io.delta:delta-core_2.12:0.7.0") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.config('spark.ui.port', '4050') \
.getOrCreate()
delta_dir = ('./output/delta')
spark.sql('CREATE DATABASE IF NOT EXISTS RIDES')
spark.sql('''
CREATE TABLE IF NOT EXISTS RIDES.YELLOW_TAXI(
VendorID long,
tpep_pickup_datetime timestamp,
tpep_dropoff_datetime timestamp,
passenger_count double,
trip_distance double,
RatecodeID double,
store_and_fwd_flag string,
PULocationID long,
DOLocationID long,
payment_type long,
fare_amount double,
extra double,
mta_tax double,
tip_amount double,
tolls_amount double,
improvement_surcharge double,
total_amount double,
congestion_surcharge double,
airport_fee double
) USING DELTA
LOCATION "{0}"
'''.format(delta_dir)
)
df = spark.read.parquet('yellow_tripdata_2021-01.parquet')
def update():
spark.sql("""MERGE INTO RIDES.YELLOW_TAXI
USING load_table
ON RIDES.YELLOW_TAXI.VendorID = load_table.VendorID and
RIDES.YELLOW_TAXI.tpep_pickup_datetime = load_table.tpep_pickup_datetime and
RIDES.YELLOW_TAXI.tpep_dropoff_datetime = load_table.tpep_dropoff_datetime and
RIDES.YELLOW_TAXI.PULocationID = load_table.PULocationID and
RIDES.YELLOW_TAXI.DOLocationID = load_table.DOLocationID
WHEN NOT MATCHED THEN
INSERT (VendorID,
tpep_pickup_datetime,
tpep_dropoff_datetime,
passenger_count,
trip_distance,
RatecodeID,
store_and_fwd_flag,
PULocationID,
DOLocationID,
payment_type,
fare_amount,
extra,
mta_tax,
tip_amount,
tolls_amount,
improvement_surcharge,
total_amount,
congestion_surcharge,
airport_fee) VALUES (VendorID,
tpep_pickup_datetime,
tpep_dropoff_datetime,
passenger_count,
trip_distance,
RatecodeID,
store_and_fwd_flag,
PULocationID,
DOLocationID,
payment_type,
fare_amount,
extra,
mta_tax,
tip_amount,
tolls_amount,
improvement_surcharge,
total_amount,
congestion_surcharge,
airport_fee)
""")
And then I load the files into load_table and run the update. When I run
from delta.tables import *
deltaTable = DeltaTable.forPath(spark, './output/delta')
fullHistoryDF = deltaTable.history()
fullHistoryDF.show()
I can see all the versions there, but just can't figure out how to query a specific version.