Hi all I m trying to use Kedro to develop a ML pipeline with

Question

Hi all I m trying to use Kedro to develop a ML pipeline with Spark using a Dataproc cluster in GCP I d like to load a table from Big Query in a Spark dataset how could I define that in the catalog I know that I can use plain PySpark to read the table but I d like to use the catalog Thanks

Deepyaman Datta · Answer

I haven t done this personally but if you use <https cloud google com dataproc docs tutorials bigquery connector spark example> then you should just be able to specify `file format bigquery` for your `spark SparkDataSet`

Balachandran Ponnusamy · Answer

< Sebastian Cardona Lozano> Hi Sebastian I am facing errors while running Kedro in Dataproc can you pls share the spark yml configuration to be used for dataproc the documentation says this You should modify this code to adapt it to your cluster s setup e g setting master to yarn any help on this is much appreciated

marrrcin · Answer

Maybe this repo will help you <https github com getindata kedro pyspark dataproc demo|https github com getindata kedro pyspark dataproc demo> It s a demo of Iris Kedro Pipeline running on Dataproc Batches Serverless Spark

Sebastian Cardona Lozano · Answer

Hi all I m trying to develop the pipeline of the model using Jupyter Lab IDE in the Dataproc cluster it is a small cluster I followed the instructions of the documentation and I created these new files 1 `spark yml` Specifying the jar with spark bigquery connector ```spark jars ``` 2 In `src lt package name gt ` a `hooks py` ```from kedro framework hooks import hook impl from pyspark import SparkConf from pyspark sql import SparkSession class SparkHooks hook impl def after context created self context gt None Initialises a SparkSession using the config defined in project s conf folder Load the spark configuration in spark yaml using the config loader parameters = context config loader get spark spark spark conf = SparkConf setAll parameters items Initialise the spark session spark session conf = SparkSession builder appName context package name config conf=spark conf spark session = spark session conf getOrCreate spark session sparkContext setLogLevel WARN ``` 3 Updated the `HOOKS` variable in `src lt package name gt settings py` as follows ```from lt package name gt hooks import SparkHooks HOOKS = SparkHooks ``` 4 In the `catalog yml` I specified the Big Query table as follows ```master table type spark SparkDataSet filepath gcp project name bigquery dataset table name in big query file format bigquery``` When I load the table from the catalog in a Jupyter Notebook and using Python I get this ```df = catalog load master table 02 25 23 03 09 47 INFO Loading data from master table SparkDataSet data catalog py 343 Traceback most recent call last opt conda miniconda3 envs py39 599 lib python3 9 site packages kedro io core py 186 in load 183 self logger debug Loading s str self 184 185 try 186 return self load 187 except DataSetError 188 raise 189 except Exception as exc opt conda miniconda3 envs py39 599 lib python3 9 site packages kedro extras datasets spark spar k dataset py 392 in load 389 if self schema 390 read obj = read obj schema self schema 391 392 return read obj load load path self file format self load args 393 394 def save self data DataFrame gt None 395 save path = strip dbfs prefix self fs prefix + str self get save path opt conda miniconda3 envs py39 599 lib python3 9 site packages pyspark sql readwriter py 177 in load 174 self schema schema 175 self options options 176 if isinstance path str 177 return self df self jreader load path 178 elif path is not None 179 if type path = list 180 path = path type ignore list item opt conda miniconda3 envs py39 599 lib python3 9 site packages py4j java gateway py 1321 in call 1318 proto END COMMAND PART 1319 1320 answer = self gateway client send command command 1321 return value = get return value 1322 answer self gateway client self target id self name 1323 1324 for temp arg in temp args opt conda miniconda3 envs py39 599 lib python3 9 site packages pyspark sql utils py 190 in deco 187 def capture sql exception f Callable Any gt Callable Any 188 def deco a Any kw Any gt Any 189 try 190 return f a kw 191 except Py4JJavaError as e 192 converted = convert exception e java exception 193 if not isinstance converted UnknownException opt conda miniconda3 envs py39 599 lib python3 9 site packages py4j protocol py 326 in get return value 323 type = answer 1 324 value = OUTPUT CONVERTER type answer 2 gateway client 325 if answer 1 == REFERENCE TYPE 326 raise Py4JJavaError 327 An error occurred while calling 0 1 2 n 328 format target id name value 329 else Py4JJavaError An error occurred while calling o243 load com google cloud spark bigquery repackaged com google cloud bigquery BigQueryException Malformed project resource name projects home bbog gd 599 profundizacion ml bdb gcp pr ac ba Expected projects lt project id gt ``` Nevertheless if I use the manually method to initialize the Spark Session and load the data It works ``` Initialize the SparkSession VER = 0 26 0 FILE NAME = f spark bigquery with dependencies 2 12 VER jar connector = f spark = SparkSession builder appName 599 produndizacion config spark jars connector getOrCreate Load data df = spark read format bigquery option table gcp project name bigquery dataset table name in big query load ```

Sebastian Cardona Lozano · Answer

Thanks for your help and sorry for the large comment slightly smiling face