Hi, I'm trying to come up with a way of using a Cloudera Spark cluster from a Kedro project. The limitation is that the spark cluster can only be accessed via Livy (https://livy.apache.org/) and it's HDFS filesystem is exposed via WebHDFS. So the spark context is longliving, does not need to be set-up and is not directly accessible from the outside world. i.e. I can not issue a SparkSession.getOrCreate() from inside the Kedro project. See the diagram I attached. So from the kedro project I need to open a Livy Session and then from the nodes send commands over the Livy session. The idea is that the data is already on the cluster (hdfs) and that the data does not need to be exiting the cluster until the very end where the end result can be "downloaded" over webhdfs. I think I need to create this livy session in the hooks.py (equivalent to the Spark example) - my code below. However I fail to see how I can then access this livy session from the nodes? I also do not (yet) see how the catalog can contain the webhdfs:// references if that data is not really accessed inside the kedro project. Any help is much appreciated. br, Ruben Example livy code (which works):

code2 = """
df = spark.read.option("header",True) \
          .parquet("/data/sandbox/my_file.parquet")
df.printSchema()
df.show(10)
"""

with LivySession.create(LIVY_URL, kind=SessionKind.PYSPARK, auth=auth) as session:
    msg = session.run(code2)
    print(msg)

my hooks.py:

from kedro.framework.hooks import hook_impl
from kedro.config import ConfigLoader
from kedro.framework.project import settings
from kedro.framework.context import KedroContext
import os
from pathlib import Path
from livy import LivySession, SessionKind
import requests
from requests.auth import HTTPBasicAuth

LIVY_URL = '<https://knox.xxxx:8443/gateway/cdp-proxy-api/livy_for_spark3>'

ca_bundle: Path = Path("/etc/pki/tls/cert.pem")
os.environ["REQUESTS_CA_BUNDLE"] = str(ca_bundle)


class LivyHooks:
    def _get_credentials(self, key):
        conf_path = f"{self.project_path}/{settings.CONF_SOURCE}"
        conf_loader = ConfigLoader(conf_source=conf_path, env="local")
        return conf_loader.get("credentials*")[key]

    def _get_basic_auth(self):

        credentials = self._get_credentials("dev_azure")

        auth = HTTPBasicAuth(credentials['username'], credentials['password'])

        return auth

    @hook_impl
    def after_context_created(self, context) -> None:
        """Initialises a Livy Session 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())

        auth = self._get_basic_auth()        

        # Initialise the livy session
        session = LivySession.create(LIVY_URL, kind=SessionKind.PYSPARK, auth=auth)

        session.wait()
        id = session.session_id

Hi all!! Does anyone have experience setting parameters via hooks? I’m trying to set some parameters via an

after_context_created

hook where I import the KedroContext. I’m modifying the KedroContext parameters property but the changes are not reflected in the parameters. I’ve seen in the documentation that the “params” property is ReadOnly, however also the

config_loader

property should be read only but the changes that I implement in the hook are propagated through the run. Any help is highly appreciated! Thanks!!!

Hi team, I am working to create a node in Kedro that saves all the excel files within each subfolders in a single excel file with different tabs. The tab names should be the subfolder_filename. The number of subfolders can be variable. Wanted to check if someone did implement this before?

Hello everyone, Using

OmegaConfigLoader

and passing parameters through

extra_params

inside

KedroSession.create

gives me an error:

InterpolationKeyError: Interpolation key 'test_ids' not found                                                                                                                                                              
                                 full_key: query_test_recommender_to_filter.sql                                                                                                                                                                         
                                 object_type=dict

The given

extra_params

are a set of strings, which I use in the

catalog.yml

to parameterize a SQL query. Does anyone know how to help me? Thanks in advance

Hi team, we are using Kedro and have two question and wonder if we can receive any help here: 1) While Kedro is trying to process one year worth of data - Reading from SQL Server and writing back to Blob Storage, it fails with following error - DatasetError: Failed while saving data to data set CSVDataSet(filepath=<path>/history.csv, protocol={'sep': |}, save_args={'index': False, 'sep': |}). Failed to upload blockTimeout on reading data from socket! 2) When we deployed Kedro to Databricks, Kedro is default looking for Local folder for configuration and since local folder is sensitive information where our credentials are kept, we are not sending it to databricks. Kedro if no local folder found then suppose to look for Base folder but erroring out with following message on databricks - Given configuration path either does not exist or is not a valid directory: /project/conf/local Any help would be greatly appreciated 🙂

Hello 👋 , Is it possible to use OmegaConf nested interpolation in the catalog? I would like to do something like:

"{table}_ds":
  type: my_custom_ds
  site: "${globals:site}"
  file_id: "${globals:file_id[${table}]}"

and in

globals.yml

file_id:
   users: ""
   items: ""

site: ""

Thanks!

Hi Team, a small doubt in kedro spark dataset:

```weather_cleaned:

type: spark.SparkDataSet

filepath: data/02_intermediate/data.parquet

file_format: parquet```

Is it possible to pass multiple fikepaths in the filepath arguement.

Hello Kedro Team, I am using SQLTAbleDataSet to save data to my DB, but in my pipeline if I am use the same variable to send it to next node, Then it loads the data again from DB instead of using it from MemoryDataSet. catalog.yml:

"{NAMESPACE}.ballbyball_final":

type: pandas.SQLTableDataSet

table_name: MODEL_{NAMESPACE}_BALLBYBALL_V2

credentials: db2

save_args:

if_exists: replace

chunksize: 10000

pipelines.py:

node(

func=total_balls_done,

inputs=["ballbyball_final_1","params:MIN_TOTAL_BALLS_MATCH"],

outputs="ballbyball_final",

---> Data is saved to DB over here

name="total_balls_done",tags="ballbyball_preprocessing"

),

node(

func=lambda x: x,

inputs="ballbyball_final",

----> Data is loaded from SQLTableDataSet instead of MemoryDataSet

outputs="cache_ballbyball_final",

name="cache_ballbyball_final",tags="ballbyball_preprocessing"

),

This is how it looks when pipeline is running:

image.png

hi team please guide me on one thing, does kedro has the facility to run the pipeline through client. i mean suppose we have deployed kedro somewhere else and in that kedro we have some pipelines too, so can we run those pipelines through our python code (from other server) via some kind of SDKs or REST calls ?

Hi! I'm using Kedro on Databricks, and using ManagedTableDataSet in the catalog to load the data I need. So far is working good, but for my use case I would like to load using a sql query. So insted of loading tableA I can load something like *select * from tableA where date <= parameter* Any suggestions on how to achieve this? https://docs.kedro.org/en/stable/kedro_datasets.databricks.ManagedTableDataset.html#kedro_datasets.databricks.ManagedTableDataset

Hi! I'm running a kedro project on databricks, and when I load the kedro project, it never shows the input cell to select the y/N option. Anyone with this problem? (I'm using runtime 11.3 LTS ML)

Hey Kedro, I'm new to ML Orchestration. I am planning to do a RAG LLM and will be processing/cleaning and embedding data for the RAG LLM without any training or fine tuning. Is Kedro used only for the ingestion and processing of data? Can it be used for embedding into a vector database like milvus or knowledge graph like neo4j?

Hi, I have a pipeline that uses a lot of memory. I profiled it with

memory_profiler

, and there is a huge memory usage, before the first node runs. This is critical for me because I am running it on a kubernetes pods, and if I use too much memory, the pod gets killed. Attached is the profile of a

dummy

pipeline that I created that prints a parameter and does nothing else. We can see that it uses nearly 200MiB of memory. I am not sure what I should look for to improve this memory usage. Is it the catalog? If so, how can I reduce it? Is it something else from kedro? Is it outside of kedro?

what's the best way of having a variable that contains the absolute path of the project root? something like

dataset:
  filepath: ${custom_kedro_resolver_magic:project_root}/data/01_raw/iris.csv

? has anybody come up with a custom resolver that can do this dynamically, rather than having to, say, hardcode it in

globals.yml

?

Maybe relying on

package_name

by getting it’s path with something like

Import package_name

Os.path.dirname(os.path.dirname(package_name.__file__))

Since

package_name

is included in pythonpath when

kedro run

, this will dynamically return package (or project) path location

Hey Everyone! I am trying to load the configuration dict in the code (specifically in pipelines.py to build a dynamic pipeline). I see this suggestion on how to do this: https://docs.kedro.org/en/stable/configuration/parameters.html#how-to-load-parameters-in-code However, this seems not to load the runtime CLI parameters at all. When looking at the OmegaConfigLoader I can see that it has by default _*runtime_params=None*_ so naturally the snippet from that link will ignore the runtime params. Is there a way so that I could keep the runtime_params alive while loading the OmegaConfigLoader directly? Like in the example? I know that the Kedro context holds the runtime_params passed in the CLI but this is not available in the pipeline.py module, right? Any suggestions?

Hello! I am trying to modify a node's input using a partial object. I am encountering the exact problem described her: https://github.com/kedro-org/kedro/issues/557 kedro version: 0.18.12 python version: 3.8.17 Could someone please help me with this, happy to give more context if the link issue is not sufficiant. Thanks in advance!

Hello, I have a question about pytest import error: This is my project's structure:

src
│   ...
└───<package_name>
│   └───pipelines
│       └───data_engineering
|           └───nodes
│               │   node_X.py
└───tests
│   └───pipelines
│       └───data_engineering
|           └───nodes
│               │   test_node_X.py

In my test file, I have to add the first 2 lines else I will get an Import error:

# Uncommenting the following 2 lines solves the error
# import sys
# sys.path.append("path_to/src")

import pytest

from <package_name>.pipelines.data_engineering.nodes.node_X import (
    func_X,
)

Is there a cleaner way than having to append the path for each test file? ERROR Message:

==================================== ERRORS ======================================================================================
________________________________________________ ERROR collecting src/tests/pipelines/data_engineering/nodes/test_node_X.py ________________________________________________
ImportError while importing test module 'path_to/src/tests/pipelines/data_engineering/nodes/test_node_X.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/home/ec2-user/miniconda3/envs/python310/lib/python3.10/importlib/__init__.py:126: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
src/tests/pipelines/data_engineering/nodes/test_node_X.py:7: in <module>
    from <package_name>.pipelines.data_engineering.nodes.node_X import (
E   ModuleNotFoundError: No module named '<package_name>'

Thank you.

Hey team, Is there a standard way to deal with possibly non-existent files in Kedro's catalog (taking a very generic use case of a node that accepts an optional argument of type dataframe)? Thank you!

Question on kedro starters: If I wanted to create a set of templates, with some functionality being common across them (like custom datasets, say - or full pipelines) - what are some approaches to keep this maintainable that work better than just copy-pasting the code between the starters?

Hey everyone, is it possible to call the Kedro API and run Kedro pipelines from a non-kedro project?

Hi all, While saving a spark parquet versioned dataset i am hitting the following error,

VersionNotFoundError: Did not find any versions for SparkDataset(file_format=parquet, filepath=<s3://bucket/folder/file_name>, 
load_args={'header': True}, save_args={'mode': overwrite}, version=Version(load=None, save='2023-12-06T13.06.41.920Z'))

config file:

_pq: &_pq
  type: spark.SparkDataSet
  file_format: parquet
  versioned: True
  load_args:
    header: True
  save_args:
    mode: overwrite

model_input.narrow_master.narrow_master:
  filepath: ${base}/model_input/master
  <<: *_pq

kedro                            0.18.14
kedro-datasets                   1.8.0

Kedro Versions, What Am I doing wrong here

Hey everyone, There is some way to get latest version of a dataset? I'm currently getting the version in a hook trigged when catalog is created, but, this only works if I explicitly inform the load version in kedro run command. I would like to get the version of a loaded dataset even when the version is the latest.

Hi maybe I missing something. I am trying to use dataset factories on kedro catalog. I need to everytime a output ends in

_query

(pattern

"{name}_query"

) to be save as text on

text.TextDataSet

on

data/{name}_query.txt

, the code on the catalog is

"{name}_query":
  type: text.TextDataSet
  filepath: data/{name}_query.txt

and on the pipeline

Hi, I'm loading and saving a databricks Delta Table using databricks.ManagedTableDataset. When I load it as a spark dataframe and save it as the same, I don't have any problem. However, when I save it as a pandas dataframe, I'm getting an error while saving the data. Any helps on why is this happening? I'm using python 3.8.10 and kedro version 0.18.14, Databricks runtime 10.4 Solved: Just upgrade the Databricks runtime to runtime 13.x https://stackoverflow.com/questions/75926636/databricks-issue-while-creating-spark-data-frame-from-pandas

I have a node with a large run time, and I am thinking about running that particular node with PyPy. I am not aware of a supported way to do this. So here is what I am thinking. I could ensure that the required inputs are saved in the Kedro data folder having been generated in earlier steps of the analysis. Then I could have a kedro node that uses subprocess.run to call PyPy on a script that performs the computationally-expensive steps and writes an intermediate output file. Finally, the kedro node loads that output back into memory and returns it (and hence can be a registered outupt in the catalog). Something resembling this:

import subprocess

PYPY_PATH = ...
SCRIPT_PATH = ...

def run_pypy_script(input_data):
    write_to_temp(input_data)
    pypy_cmd = [PYPY_PATH, SCRIPT_PATH]
    subprocess.run(pypy_cmd, check=True)
    return load_results()

Any thoughts on this approach, or a better approach? https://docs.python.org/3/library/subprocess.html#subprocess.run

Are there any commercial products that integrate kedro into their offering?

Hi, I have upgraded to kedro==0.18.14 from 0.18.1. The logging behaviour is unexpected in 0.18.14. Normally in 0.18.1 if the code breaks, it allows to click on the line of the code where it broke which takes to the specific line in the project. Unfortunately, in 0.18.14 it shows the text in the console but it is not clickable. The traceback is as shown below. and the loggin-yml is like

version: 1

disable_existing_loggers: False

formatters:
  simple:
    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

handlers:
  console:
    class: logging.StreamHandler
    level: INFO
    formatter: simple
    stream: <ext://sys.stdout>

  info_file_handler:
    class: logging.handlers.RotatingFileHandler
    level: INFO
    formatter: simple
    filename: info.log
    maxBytes: 10485760 # 10MB
    backupCount: 20
    encoding: utf8
    delay: True

  error_file_handler:
    class: logging.handlers.RotatingFileHandler
    level: ERROR
    formatter: simple
    filename: errors.log
    maxBytes: 10485760 # 10MB
    backupCount: 20
    encoding: utf8
    delay: True

  rich:
    class: kedro.logging.RichHandler
    rich_tracebacks: True
    # Advance options for customisation.
    # See <https://docs.kedro.org/en/stable/logging/logging.html#project-side-logging-configuration>
    # tracebacks_show_locals: False

loggers:
  kedro:
    level: INFO

  epc_fi:
    level: INFO

root:
  handlers: [console, info_file_handler, error_file_handler]

Can someone help to make the traceback clickable ?

Hi ! I want to modify a small thing of the load method of a specific Kedro Dataset. Is there a way to achieve this without needing to create a new Custom Data Set?