lynx.kite module reference

Python interface for the LynxKite Remote API.

The default LynxKite connection parameters can be configured through the following environment variables:

LYNXKITE_ADDRESS=https://lynxkite.example.com/
LYNXKITE_USERNAME=user@company
LYNXKITE_PASSWORD=my_password
LYNXKITE_PUBLIC_SSL_CERT=/tmp/lynxkite.crt

Example usage:

import lynx.kite
lk = lynx.kite.LynxKite()
lk.createExampleGraph().sql('select * from scalars').df()
class lynx.kite.AtomicBox(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, operation: str, inputs: Dict[str, lynx.kite.State], parameters: Dict[str, Any], manual_box_id: str = None)

Bases: lynx.kite.Box

An AtomicBox is a Box that can not be further decomposed. It corresponds to a single frontend operation.

box_id_base() → str

The base of the box_id, which is used when we save a workspace, containing this box.

name()

Either the name in the box catalog or the name under which the box is saved.

class lynx.kite.Box(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, inputs: Dict[str, lynx.kite.State], parameters: Dict[str, Any], manual_box_id: str = None)

Bases: object

Represents a box in a workspace segment.

It can store workspace segments, connected to its input plugs.

box_id_base() → str

The base of the box_id, which is used when we save a workspace, containing this box.

is_box(operation: str) → bool

Checks if the box is the operation box.

name() → str

Either the name in the box catalog or the name under which the box is saved.

register(side_effect_collector: lynx.kite.SideEffectCollector)
to_json(id_resolver: Callable[Box, str], workspace_root: str, subworkspace_path: str) → NewType.<locals>.new_type

Creates the json representation of a box in a workspace.

The inputs have to be connected, and all the attributes have to be defined when we call this.

trigger() → None

Triggers this box.

Can be used on triggerable boxes like saveToSnapshot and export boxes.

class lynx.kite.BoxCatalog(boxes: Dict[str, types.SimpleNamespace])

Bases: object

Stores box metadata.

Offers utility functions to query box metadata information.

box_names() → List[str]
inputs(name: str) → List[str]
operation_id(name: str) → str
outputs(name: str) → List[str]
class lynx.kite.BoxPath(base: lynx.kite.Box, stack: List[lynx.kite.CustomBox] = [])

Bases: object

Represents a box (which can be inside (nested) custom boxes). It can be used for example to trigger boxes inside custom boxes.

stack[i+1] is always a box contained by the workspace referred by the custom box stack[i] and base is a box contained by stack[-1].

add_box_as_base(new_base: lynx.kite.Box) → lynx.kite.BoxPath

Takes a box inside the current base as the new base and puts the current base on the top of the stack.

add_box_as_prefix(box: lynx.kite.CustomBox) → lynx.kite.BoxPath
static dependencies(bps: Collection[BoxPath]) → Dict[lynx.kite.BoxPath, Set[lynx.kite.BoxPath]]

Returns the dependencies between the given boxes.

dependency_representative() → lynx.kite.BoxPath

Returns the path of the box that should be used in dependency calculations.

For most boxes (like SQL1) this is themselves. Some boxes (like Compute inputs) have no outputs or rarely have boxes consuming their outputs. For these boxes we use their inputs for the purposes of dependency calculations. A full example:

[Create example graph] -> [Compute inputs]
  |
  v
[SQL1]                 -> [Compute inputs]

Only the Compute boxes are triggerable here, but there is no explicit dependency between them. You could trigger the second Compute box even if the first Compute box did not exist. But we want to order the Compute boxes in the intuitive way: as if the second one depended on the first. So we use their inputs as their representatives in the dependency calculation.

parent(input_name: str) → lynx.kite.BoxPath
parents() → List[lynx.kite.BoxPath]
snatch() → lynx.kite.Box

Returns a box that is accessible from outside and whose output is the same as the that of the box referred by this BoxPath.

stack_and_base() → List[lynx.kite.Box]
to_dict()

Returns a (human readable) dict representation of this object.

to_string_id(outer_ws) → str

Can be used in automation, to generate unique task ids.

stack[0] is supposed to be in the outer_ws workspace.

class lynx.kite.CustomBox(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, workspace: lynx.kite.Workspace, inputs: Dict[str, lynx.kite.State], parameters: Dict[str, Any], manual_box_id: str = None)

Bases: lynx.kite.Box

A CustomBox is a Box composed of multiple other boxes.

box_id_base()

The base of the box_id, which is used when we save a workspace, containing this box.

name()

Either the name in the box catalog or the name under which the box is saved.

snatch(box: lynx.kite.Box) → lynx.kite.Box

Takes a box that is inside the workspace referred to by this custom box and returns an equivalent box that is accessible from outside.

class lynx.kite.DataFrameSender(lk)

Bases: object

Class to send some dataframe to LynxKite in Parquet format. We do this by saving the dataframe to a local Parquet file and then upload it to LynxKite.

save_dataframe_to_local_file(df, tmp_dir) → str

Saves the dataframe as a single Parquet file in tmpdir Returns the actual path of the binary that was written.

send(df)

Sends the local Parquet file to LynxKite

class lynx.kite.ExternalComputationBox(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, inputs: List[lynx.kite.State], parameters: Dict[str, Any], fn: Callable, args: inspect.BoundArguments, manual_box_id: str = None)

Bases: lynx.kite.SingleOutputAtomicBox

A box that runs external computation when triggered. Use it via @external.

class lynx.kite.InputTable(lk, lk_table: lynx.kite._DownloadableLKTable)

Bases: object

Input tables for external computations (@external) are translated to these objects.

lk() → lynx.kite.State

Returns a LynxKite State.

pandas()

Returns a Pandas DataFrame.

spark(spark)

Takes a SparkSession as the argument and returns the table as a Spark DataFrame.

exception lynx.kite.LynxException(error)

Bases: Exception

Raised when LynxKite indicates that an error has occured while processing a command.

class lynx.kite.LynxKite(username: str = None, password: str = None, address: str = None, certfile: str = None, oauth_token: str = None, signed_token: str = None, box_catalog: lynx.kite.BoxCatalog = None)

Bases: object

A connection to a LynxKite instance.

Some LynxKite API methods take a connection argument which can be used to communicate with multiple LynxKite instances from the same session. If no arguments to the constructor are provided, then a connection is created using the following environment variables: LYNXKITE_ADDRESS, LYNXKITE_USERNAME, LYNXKITE_PASSWORD, LYNXKITE_PUBLIC_SSL_CERT, LYNXKITE_OAUTH_TOKEN, LYNXKITE_SIGNED_TOKEN.

address() → str
box_catalog() → lynx.kite.BoxCatalog
certfile() → Optional[str]
change_acl(file: str, readACL: str, writeACL: str)

Sets the read and write access control list for a path in LynxKite.

clean_file_system()

Deletes the data files which are not referenced anymore.

create_dir(path: str, privacy: str = 'public-read')
download_file(path: str) → bytes
empty_cleaner_trash()

Empties the cleaner trash.

export_box(outputs: Dict[Tuple[str, str], types.SimpleNamespace], box_id: str) → types.SimpleNamespace

Equivalent to triggering the export. Returns the exportResult output.

export_operation_names() → List[str]

When we use an export operation instead of an export box, the export will be triggered automatically.

fetch_states(boxes: List[NewType.<locals>.new_type], parameters: Dict = {}) → Dict[Tuple[str, str], types.SimpleNamespace]
fetch_workspace_output_states(ws: lynx.kite.Workspace, save_under_root: str = None) → Dict[Tuple[str, str], types.SimpleNamespace]
from_df(df, index=True)

Converts a pandas dataframe to a LynxKite table.

Set index to False if you don’t want to include the index. Warning: the index setting feature only works with pandas>=0.24.0

get_data_files_status()

Returns the amount of space used by LynxKite data, various cleaning methods and the amount of space they can free up.

get_directory_entry(path: str)

Returns details about a LynxKite path. The returned object has the following fields: exists, isWorkspace, isSnapshot, isDirectory

get_export_result(state: str) → types.SimpleNamespace
get_parquet_metadata(path: str)

Reads the metadata of a parquet file and returns the number of rows.

get_prefixed_path(path: str)

Resolves a path on a distributed file system. The path has to be specified using LynxKite’s prefixed path syntax. (E.g. DATA$/my_file.csv.)

The returned object has an exists and a resolved attribute. resolved is a string containing the absolute path.

get_project(state: str, path: str = '') → types.SimpleNamespace
get_scalar(guid: str) → types.SimpleNamespace
get_state_id(state: lynx.kite.State) → str
get_table_data(state: str, limit: int = -1) → types.SimpleNamespace
get_workspace(path: str, stack: List[str] = []) → types.SimpleNamespace
get_workspace_boxes(path: str, stack: List[str] = []) → List[types.SimpleNamespace]
home() → str
import_box(boxes: List[NewType.<locals>.new_type], box_id: str) → List[NewType.<locals>.new_type]

Equivalent to clicking the import button for an import box. Returns the updated boxes.

import_operation_names() → List[str]

When we use an import operation instead of an import box, “run import” will be triggered automatically.

list_dir(dir: str = '') → List[types.SimpleNamespace]

List the objects in a directory, with their names, types, notes, and other optional data about projects.

move_to_cleaner_trash(method: str)

Moves LynxKite data files specified by the cleaning method into the cleaner trash. The possible values of method are defined in the result of get_data_files_status.

oauth_token() → Optional[str]
operation_names() → List[str]
password() → Optional[str]
recursively_collect_customboxes(ws: lynx.kite.Workspace, path) → Set[Tuple[str, lynx.kite.CustomBox]]
remove_name(name: str, force: bool = False)

Removes an object named name.

save_snapshot(path: str, stateId: str)
save_workspace(path: str, boxes: List[NewType.<locals>.new_type], overwrite: bool = True)
save_workspace_recursively(ws: lynx.kite.Workspace, save_under_root: str = None) → Tuple[str, str]
set_cleaner_min_age(days: float)
set_executors(count)
signed_token() → Optional[str]
sql(sql: str, *args, **kwargs) → lynx.kite.Box

Shorthand for sql1, sql2, …, sql10 boxes.

Use with positional arguments to go with the default naming, like:

lk.sql('select name from one join two', state1, state2)

Or pass the states in keyword arguments to use custom input names:

lk.sql('select count(*) from people', people=state1)

In either case you can pass in extra keyword arguments which will be passed on to the SQL boxes.

trigger_box(workspace_name: str, box_id: str, custom_box_stack: List[str] = [])

Triggers the computation of all the GUIDs in the box which is in the saved workspace named workspace_name and has boxID=box_id. If custom_box_stack is not empty, it specifies the sequence of custom boxes which leads to the workspace which contains the box with the given box_id.

upload(data: bytes, name: str = None)

Uploads a file that can then be used in import methods.

prefixed_path = lk.upload(‘id,namen1,Bob’)

uploadCSVNow(data: bytes, name: str = None)

Uploads CSV data and returns a table state.

uploadParquetNow(data: bytes, name: str = None)

Uploads Parquet file and returns a table state.

username() → str
workspace(name: str = None, parameters: List[lynx.kite.WorkspaceParameter] = [], inputs: List[str] = None) → Callable[Callable[..., Dict[str, lynx.kite.State]], lynx.kite.Workspace]
workspace_with_side_effects(name: str = None, parameters: List[lynx.kite.WorkspaceParameter] = [], inputs: List[str] = None) → Callable[Callable[..., Dict[str, lynx.kite.State]], lynx.kite.Workspace]
class lynx.kite.PandasDataFrameSender(lk)

Bases: lynx.kite.DataFrameSender

save_dataframe_to_local_file(df, path) → str

Saves the dataframe as a single Parquet file in tmpdir Returns the actual path of the binary that was written.

class lynx.kite.ParametricParameter(parametric_expr: str)

Bases: object

Represents a parametric parameter value. The short alias pp is defined for this type.

class lynx.kite.PizzaBox

Bases: lynx.kite.LynxKite

class lynx.kite.Placeholder(value=None)

Bases: object

Universal placeholder. Use it whenever you need to hold a place.

class lynx.kite.SideEffectCollector

Bases: object

AUTO = <lynx.kite.SideEffectCollector object>
add_box(box: lynx.kite.Box) → None
all_triggerables() → Iterable[lynx.kite.BoxPath]
class lynx.kite.SingleOutputAtomicBox(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, operation: str, inputs: Dict[str, lynx.kite.State], parameters: Dict[str, Any], output_name: str, manual_box_id: str = None)

Bases: lynx.kite.AtomicBox, lynx.kite.State

An AtomicBox with a single output. This makes chaining multiple operations after each other possible.

class lynx.kite.SingleOutputCustomBox(box_catalog: lynx.kite.BoxCatalog, lk: lynx.kite.LynxKite, workspace: lynx.kite.Workspace, inputs: Dict[str, lynx.kite.State], parameters: Dict[str, Any], output_name: str, manual_box_id: str = None)

Bases: lynx.kite.CustomBox, lynx.kite.State

An CustomBox with a single output. This makes chaining multiple operations after each other possible.

class lynx.kite.SparkDataFrameSender(lk)

Bases: lynx.kite.DataFrameSender

save_dataframe_to_local_file(df, target_dir) → str

Saves the dataframe as a single Parquet file in tmpdir Returns the actual path of the binary that was written.

class lynx.kite.State(box: lynx.kite.Box, output_plug_name: str)

Bases: object

Represents a named output plug of a box.

It can recursively store the boxes which are connected to the input plugs of the box of this state.

columns()

Returns a list of columns if this state is a table.

compute() → None

Triggers the computation of this state.

Uses a temporary folder to save a temporary workspace for this computation.

df(limit: int = -1)

Returns a Pandas DataFrame if this state is a table.

get_progress()

Returns progress info about the state.

get_project() → types.SimpleNamespace

Returns the project metadata if this state is a project.

get_table_data(limit: int = -1) → types.SimpleNamespace

Returns the “raw” table data if this state is a table.

operation_names()
persist() → lynx.kite.SingleOutputAtomicBox

Same as x.sql('select * from input', persist='yes').

run_export() → str

Triggers the export if this state is an exportResult.

Returns the prefixed path of the exported file. This method is deprecated, only used in tests, where we need the export path.

save_snapshot(path: str) → None

Save this state as a snapshot under path.

save_to_sequence(tss, date: datetime.datetime) → None

Save this state to the tss TableSnapshotSequence with date as the date of the snapshot.

sql(sql: str, **kwargs) → lynx.kite.SingleOutputAtomicBox
class lynx.kite.Workspace(terminal_boxes: List[lynx.kite.Box] = [], name: Optional[str] = None, custom_box_id_base: Optional[str] = None, side_effect_paths: List[lynx.kite.BoxPath] = [], input_boxes: List[lynx.kite.AtomicBox] = [], ws_parameters: List[lynx.kite.WorkspaceParameter] = [])

Bases: object

Immutable class representing a LynxKite workspace.

custom_boxes() → List[lynx.kite.CustomBox]
find(box_id_base: str) → lynx.kite.BoxPath

Returns the BoxPath for the box nested in the workspace whose box_id_base is the given string.

Raises an error if there is not exactly one such a box.

find_all(box_id_base: str) → List[lynx.kite.BoxPath]

Returns the BoxPaths for all boxes nested in the workspace whose box_id_base is the given string.

has_date_parameter() → bool
has_local_date_parameter() → bool
id_of(box: lynx.kite.Box) → str
safename() → str
save(saved_under_folder: str) → str
side_effect_paths() → List[lynx.kite.BoxPath]
terminal_box_ids() → List[str]
to_json(workspace_root: str, subworkspace_path: str) → List[NewType.<locals>.new_type]
trigger(box_to_trigger: lynx.kite.BoxPath)

Triggers one side effect.

Assumes the workspace is not saved, so saves it under a random folder.

trigger_all_side_effects()

Triggers all side effects.

Also saves the workspace under a temporary folder.

trigger_saved(box_to_trigger: lynx.kite.BoxPath, saved_under_folder: str)

Triggers one side effect.

Assumes the workspace is saved under saved_under_root.

class lynx.kite.WorkspaceParameter(name: str, kind: str, default_value: str = '')

Bases: object

Represents a workspace parameter declaration.

to_json() → Dict[str, str]
lynx.kite.dedent(s: Union[str, ParametricParameter]) → Union[str, lynx.kite.ParametricParameter]

Format sql queries to look nicer on the UI.

lynx.kite.escape(s: Union[str, ParametricParameter]) → Union[str, lynx.kite.ParametricParameter]

Sanitizes a string for injecting into a generated SQL query.

lynx.kite.external(fn: Callable)

Decorator for executing computation outside of LynxKite in a LynxKite workflow.

Returns a custom box that internally exports the input tables and runs the external computation on them when it is triggered. The output of this custom box is the result of the external computation as a LynxKite table.

Example:

@external
def titled_names(table, default):
  df = table.pandas()
  df['gender'] = df.gender.fillna(default)
  df['titled'] = np.where(df.gender == 'Female', 'Ms ' + df.name, 'Mr ' + df.name)
  return df

t = titled_names(lk.createExampleGraph(), 'Female')
t.trigger() # Causes the function to run.
print(t.df())

Table state parameters of the function call are exported into a Parquet file. You can read them manually, or via the methods of the InputTable objects that are passed to your function in their place:

  • filename is the Hadoop path for the file. (I.e. file:/something or hdfs:///something.)
  • pandas() loads the Parquet file into a Pandas DataFrame.
  • spark() loads the Parquet file into a PySpark DataFrame.
  • lk() loads the Parquet file into a LynxKite state.

Your function can return one of the following types:

  • pandas.DataFrame to be written to a Parquet file and imported in LynxKite.
  • spark.DataFrame to be written to a Parquet file and imported in LynxKite.
  • A string that is the LynxKite prefixed path to a Parquet file that is your output.
  • A LynxKite table state to be written to a Parquet file and imported in LynxKite.

Why use LynxKite states in external computations? This allows the use of LynxKite as in a notebook environment. You can access the actual data and write code that is conditional on the data.

lynx.kite.map_args(signature: inspect.Signature, bound: inspect.BoundArguments, map_fn: Callable[[str, Any], Any]) → inspect.BoundArguments
lynx.kite.pp

alias of lynx.kite.ParametricParameter

lynx.kite.random_filename() → str
lynx.kite.serialize_deps(deps: Dict[Any, Set[Any]]) → List[Any]

Returns the keys of deps in an execution order that respects the dependencies.

lynx.kite.subworkspace(fn: Callable)

Allows using the decorated function as a LynxKite custom box.

Example use:

@subworkspace
def my_func(input1, other_arg1):
  return input1.sql1(sql=f'select {other_arg1} from vertices')

df = my_func(lk.createExampleGraph(), 'name').df()

To make a multi-output box, return a dictionary instead of a state.

my_func() can have any number of positional or keyword arguments. The arguments carrying states will be turned into the inputs of the custom box.

Use @ws_param to take workspace parameters.

To define a custom box with side-effects, take an argument with a default value of SideEffectCollector.AUTO. A fresh SideEffectCollector() will be provided each time the function is called.

@subworkspace
def my_func(input1, sec=SideEffectCollector.AUTO):
  input1.saveAsSnapshot('x').register(sec)

my_func(lk.createExampleGraph()).trigger()
lynx.kite.text(name: str, default: str = '') → lynx.kite.WorkspaceParameter

Helper function to make it easy to define a text kind ws parameter.

lynx.kite.ws_name(name: str)

Specifies the name of the wrapped subworkspace.

Example use:

@ws_name('My nice workspace')
@subworkspace
def my_func(input1):
  return input1.sql1(sql='select * from vertices')

my_func(lk.createExampleGraph())
lynx.kite.ws_param(name: str, default: str = '', description: str = '')

Adds a workspace parameter to the wrapped subworkspace.

Example use:

@ws_param('p1')
@ws_param('p2', default='x', description='The second column.')
@subworkspace
def my_func(input1):
  return input1.sql1(sql=pp('select $p1, $p2 from vertices'))

my_func(lk.createExampleGraph(), p1='age', p2='name')

Often you just want to pass your workspace parameter along unchanged. You could just write box(param=pp('$ws_param')). But if you take a keyword argument with the same name as a workspace parameter, this is even easier:

@ws_params('p1')
@subworkspace
def my_func(input1, p1):
  return input1.sql1(sql=p1)  # Equivalent to sql=pp('$p1').

Indices and tables: