sparkly package

Subpackages

Submodules

sparkly.index_config module

class sparkly.index_config.IndexConfig(*, store_vectors: bool = False, id_col: str = '_id', weighted_queries: bool = False)

Bases: object

add_concat_field(field: str, concat_fields: Iterable[str], analyzers: Iterable[str])

Add a new concat field to be indexed with this config

Parameters:

  • field (str) – The name of the field that will be added to the index

  • concat_fields (set, list or tuple of strs) – the fields in the table that will be concatenated together to create field

  • analyzers (set, list or tuple of str) – The names of the analyzers that will be used to index the field

add_field(field: str, analyzers: Iterable[str])

Add a new field to be indexed with this config

Parameters:

  • field (str) – The name of the field in the table to the index

  • analyzers (set, list or tuple of str) – The names of the analyzers that will be used to index the field

freeze()
Returns:

a frozen deepcopy of this index config

Return type:

IndexConfig

classmethod from_json(data)

construct an index config from a dict or json string, see IndexConfig.to_dict for expected format

Return type:

IndexConfig

get_analyzed_fields(query_spec=None)

Get the fields used by the index or query_spec. If query_spec is None, the fields that are used by the index are returned.

Parameters:

query_spec (QuerySpec, optional) – if provided, the fields that are used by query_spec in creating a query

Returns:

the fields used

Return type:

list of str

property id_col

The unique id column for the records in the index this must be a 32 or 64 bit integer

property is_frozen

returns: True if this index is frozen (not modifiable) else False :rtype: bool

remove_field(field: str)

remove a field from the config

Parameters:

field (str) – the field to be removed from the config

Returns:

True if the field existed else False

Return type:

bool

property store_vectors

True if the term vectors in the index should be stored, else False

to_dict()

convert this IndexConfig to a dictionary which can easily be stored as json

Returns:

A dictionary representation of this IndexConfig

Return type:

dict

to_json()

Dump this IndexConfig to a valid json strings

Return type:

str

property weighted_queries

True if the term vectors in the index should be stored, else False

sparkly.analysis module

class sparkly.analysis.Gram2Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
class sparkly.analysis.Gram3Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
class sparkly.analysis.Gram4Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
class sparkly.analysis.PythonAlnumTokenFilter(tokenStream)

Bases: PythonFilteringTokenFilter

accept()
class sparkly.analysis.StandardEdgeGram36Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
class sparkly.analysis.StrippedGram3Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
initReader(fieldName, reader)
class sparkly.analysis.UnfilteredGram3Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
class sparkly.analysis.UnfilteredGram5Analyzer

Bases: PythonAnalyzer

createComponents(fieldName)
sparkly.analysis.analyze(analyzer, text, with_offset=False)

Apply the analyzer to the text and return the tokens, optionally with offsets

Parameters:

  • analyzer – The lucene analyzer to be applied

  • text (str) – the text that will be analyzer

  • with_offset (bool) – if true, return the offsets with the tokens in the form (TOKEN, START_OFFSET, END_OFFSET)

Returns:

a list of tokens potentially with offsets

Return type:

list of str or tuples

sparkly.analysis.analyze_generator(analyzer, text, with_offset=False)

Apply the analyzer to the text and return the tokens, optionally with offsets

Parameters:

  • analyzer – The lucene analyzer to be applied

  • text (str) – the text that will be analyzer

  • with_offset (bool) – if true, return the offsets with the tokens in the form (TOKEN, START_OFFSET, END_OFFSET)

Returns:

a list of tokens potentially with offsets

Return type:

generator of str or tuples

sparkly.analysis.get_shingle_analyzer()
sparkly.analysis.get_standard_analyzer_no_stop_words()

sparkly.search module

class sparkly.search.Searcher(index: Index, search_chunk_size: Annotated[int, Gt(gt=0)] = 500)

Bases: object

class for performing bulk search over a dataframe

get_full_query_spec()

get a query spec that searches on all indexed fields

search(search_df: DataFrame, query_spec: QuerySpec, limit: Annotated[int, Gt(gt=0)], id_col: str = '_id')

perform search for all the records in search_df according to query_spec

Parameters:

  • search_df (pyspark.sql.DataFrame) – the records used for searching

  • query_spec (QuerySpec) – the query spec for searching

  • limit (int) – the topk that will be retrieved for each query

  • id_col (str) – the id column from search_df that will be output with the query results

Returns:

a pyspark dataframe with the schema (id2, id1_list array<long> , scores array<float>, search_time float)

Return type:

pyspark DataFrame

sparkly.search.search(index, query_spec, limit, search_recs)
sparkly.search.search_gen(index, query_spec, limit, search_recs)

sparkly.utils module

class sparkly.utils.Timer

Bases: object

utility class for timing execution of code

get_interval()

get the time that has elapsed since the object was created or the last time get_interval() was called

Return type:

float

get_total()

get total time this Timer has been alive

Return type:

float

set_start_time()

set the start time to the current time

sparkly.utils.atomic_unzip(zip_file_name, output_loc)

atomically unzip the file, that is this function is safe to call from multiple threads at the same time

Parameters:

  • zip_file_name (str) – the name of the file to be unzipped

  • output_loc (str) – the location that the file will be unzipped to

sparkly.utils.attach_current_thread_jvm()

attach the current thread to the jvm for PyLucene

sparkly.utils.auc(x)
sparkly.utils.check_tables_auto(table_a: DataFrame | DataFrame, id_col_table_a: str, table_b: DataFrame | DataFrame, id_col_table_b: str)

Check that table_a and table_b have valid id columns. Check that table_b columns are a supserset of table_a columns.

Parameters:

  • table_a (Union[pd.DataFrame, sql.DataFrame]) – The table A to be indexed.

  • id_col_table_a (str) – The column name of the id column in table A.

  • table_b (Union[pd.DataFrame, sql.DataFrame]) – The table B to be searched.

  • id_col_table_b (str) – The column name of the id column in table B.

Raises:

  • ValueError – If table_a or table_b do not have a valid id column.

  • ValueError – If table_b columns are not a supserset of table_a columns.

sparkly.utils.check_tables_manual(table_a: DataFrame | DataFrame, id_col_table_a: str, table_b: DataFrame | DataFrame, id_col_table_b: str)

Check that table_a and table_b have valid id columns.

Parameters:

  • table_a (Union[pd.DataFrame, sql.DataFrame]) – The table A to be indexed.

  • id_col_table_a (str) – The column name of the id column in table A.

  • table_b (Union[pd.DataFrame, sql.DataFrame]) – The table B to be searched.

  • id_col_table_b (str) – The column name of the id column in table B.

Raises:

ValueError – If table_a or table_b do not have a valid id column.

sparkly.utils.get_index_name(n, *postfixes)

utility function for generating index names in a uniform way

sparkly.utils.get_logger(name, level=10)

Get the logger for a module

Return type:

Logger

sparkly.utils.init_jvm(vmargs=[])

initialize the jvm for PyLucene

Parameters:

vmargs (list[str]) – the jvm args to the passed to the vm

sparkly.utils.invoke_task(task)

invoke a task created by joblib.delayed

sparkly.utils.is_null(o)

check if the object is null, note that this is here to get rid of the weird behavior of np.isnan and pd.isnull

sparkly.utils.is_persisted(df)

check if the pyspark dataframe is persist

sparkly.utils.kill_loky_workers()

kill all the child loky processes of this process. used to prevent joblib from sitting on resources after using joblib.Parallel to do computation

sparkly.utils.local_parquet_to_spark_df(file)
sparkly.utils.norm_auc(x)
sparkly.utils.persisted(df, storage_level=StorageLevel(True, True, False, False, 1))

context manager for presisting a dataframe in a with statement. This automatically unpersists the dataframe at the end of the context

sparkly.utils.repartition_df(df, part_size, by=None)

repartition the dataframe into chunk of size ‘part_size’ by column ‘by’

sparkly.utils.spark_to_pandas_stream(df, chunk_size, by='_id')

repartition df into chunk_size and return as iterator of pandas dataframes

sparkly.utils.type_check(var, var_name, expected)

type checking utility, throw a type error if the var isn’t the expected type

sparkly.utils.type_check_iterable(var, var_name, expected_var_type, expected_element_type)

type checking utility for iterables, throw a type error if the var isn’t the expected type or any of the elements are not the expected type

sparkly.utils.zip_dir(d, outfile=None)

Zip a directory d and output it to outfile. If outfile is not provided, the zipped file is output in /tmp

Parameters:

  • d (str or Path) – the directory to be zipped

  • outfile (str or Path, optional) – the output location of the zipped file

Returns:

the path to the new zip file

Return type:

Path

Module contents