Delta Time Travel with SQL error extraneous input '0' expecting {<EOF>, ';'}(line 1, pos 38)

Viewed 211

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.

1 Answers

The syntax is incorrect - if you're fetching the specific version, then you need to use syntax VERSION AS OF NNNN, but you're simply use AS OF NNNN. See docs for details.

Update: This syntax will be supported only starting with Spark 3.3 (See SPARK-37219 for details) that will be released in the near future.

Related