PySpark to IBIS

[vmsaipreeth]

Hi Team,

Let’s assume we have a pyspark dataframe. How easily can I convert pyspark dataframe to an ibis table or dataframe?

Thanks!

[lostmygithubaccount]

hi @vmsaipreeth, which Spark platform are you using? I think the best way would be to save the data as a table via PySpark, then access that table through con.table("table_name", where the con is an Ibis PySpark backend connection

alternatively, you could use PyArrow to ingest the data from the PySpark dataframe into an Ibis table

[vmsaipreeth]

Alright. It would be great if you can show an example on how to ingest the data from pyspark dataframe to an ibis table. TIA!

[lostmygithubaccount]

I don't have a PySpark environment setup conveniently right now, but it should look something like:

import ibis
import pyarrow as pa

# Assuming `df` is your PySpark DataFrame
pandas_df = df.toPandas()
arrow_table = pa.Table.from_pandas(pandas_df)

t = ibis.memtable(arrow_table)

I don't see a direct method for outputting a PySpark dataframe to a PyArrow table or record batch reader, though I suspect one exists. an ibis.memtable can take in a pandas dataframe directly (ibis.memtable(pandas_df) instead above) or PyArrow objects

alternatively, you could write out the PySpark dataframe to a table (native format, Delta Lake) or other file format (Parquet, etc.) and read those with the Ibis methods

[gforsyth]

hi @vmsaipreeth -- do you have an existing Spark session that this DataFrame is attached to?

Ibis connects to a Spark Session, so if the DataFrame is a view in that session, you can interact with it that way. It would probably look something like:

import ibis
from pyspark.sql import SparkSession

session = SparkSession.builder.appName("my_session").getOrCreate()

existing_spark_df.createOrReplaceTempView("my_df")

con = ibis.pyspark.connect(session)

my_df = con.table("my_df")

[vmsaipreeth]

Hi @gforsyth
Thanks for your reply. Will give it a try. Also, just checking if this works the same with trino as well?

Like Ibis connects to a trino session:

con = ibis.trino.connect(session)
// session -> trino session

TIA!

[gforsyth]

I don't believe Trino has "sessions" in the same way as Spark does. With trino you are connecting to a server that has some number of catalogs on it. But any data that is present in a given catalog (or has been recently updated) will be visible to all other users who are connected to that same database / schema.