a bg`@sRddlmZddlmZmZmZmZmZer@ddlm Z m Z m Z GdddZ dS)) annotations) TYPE_CHECKINGAnyIterableListOptional) DataFrameRow SparkSessionc@seZdZdZd1dddddddd d Zed2d d d ddddZddddZddddZd d dddZ d3dd dddZ d d dddZ dd d!d"d#Z d$d%d&d'd(Z d4d d d d*d+d,Zd5dd dd-d.Zd6d d d d*d/d0ZdS)7SparkSQLz;SparkSQL is a utility class for interacting with Spark SQL.NzOptional[SparkSession]z Optional[str]zOptional[List[str]]int) spark_sessioncatalogschema ignore_tablesinclude_tablessample_rows_in_table_infoc Cs*zddlm}Wnty*tdYn0|r4|n|j|_|durV|jj||durl|jj|t | |_ |rt |nt |_ |j r|j |j }|rt d|d|rt |nt |_|jr|j|j }|rt d|d|} | rt | n|j |_t|ts td||_dS) aInitialize a SparkSQL object. Args: spark_session: A SparkSession object. If not provided, one will be created. catalog: The catalog to use. If not provided, the default catalog will be used. schema: The schema to use. If not provided, the default schema will be used. ignore_tables: A list of tables to ignore. If not provided, all tables will be used. include_tables: A list of tables to include. If not provided, all tables will be used. sample_rows_in_table_info: The number of rows to include in the table info. Defaults to 3. rr Fpyspark is not installed. Please install it with `pip install pyspark`Nzinclude_tables  not found in databasezignore_tables z,sample_rows_in_table_info must be an integer) pyspark.sqlr ImportErrorbuilder getOrCreate_sparkrZsetCurrentCatalogZsetCurrentDatabaseset_get_all_table_names _all_tables_include_tables ValueError_ignore_tablesget_usable_table_namesZ_usable_tables isinstancer TypeError_sample_rows_in_table_info) selfrrrrrrr missing_tablesZ usable_tablesr(u/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/langchain_community/utilities/spark_sql.py__init__ s@       zSparkSQL.__init__strzOptional[dict]r) database_uri engine_argskwargsreturncKsLzddlm}Wnty*tdYn0|j|}||fi|S)zzCreating a remote Spark Session via Spark connect. For example: SparkSQL.from_uri("sc://localhost:15002") rrr)rr rrremoter)clsr,r-r.r Zsparkr(r(r)from_uriKs  zSparkSQL.from_uriz Iterable[str])r/cCs|jr |jSt|j|jS)zGet names of tables available.)rsortedrr!)r&r(r(r)r"\szSparkSQL.get_usable_table_namescCs(|jdd}ttdd|S)Nz SHOW TABLES tableNamecSs|jSN)r4)rowr(r(r)ez/SparkSQL._get_all_table_names..)rsqlselectcollectlistmap)r&rowsr(r(r)rcszSparkSQL._get_all_table_names)tabler/cCs6|jd|dj}|d}|d|dS)NzSHOW CREATE TABLE rZUSING;)rr9r;Zcreatetab_stmtfind)r&r?Z statementZusing_clause_indexr(r(r)_get_create_table_stmtgs zSparkSQL._get_create_table_stmt) table_namesr/cCs|}|dur6t||}|r2td|d|}g}|D]D}||}|jrx|d7}|d||d7}|d7}||q>d|}|S)Nz table_names rz /* z*/z ) r"r differencer rBr%_get_sample_spark_rowsappendjoin)r&rCZall_table_namesr'ZtablesZ table_nameZ table_infoZ final_strr(r(r)get_table_infoos    zSparkSQL.get_table_infocCsd|d|j}|j|}dttdd|jj}z"||}ddd|D}Wnt yrd }Yn0|jd |d |d|S) NzSELECT * FROM z LIMIT  cSs|jSr5)name)fr(r(r)r7r8z1SparkSQL._get_sample_spark_rows..rDcSsg|]}d|qS)rJ)rH).0r6r(r(r) r8z3SparkSQL._get_sample_spark_rows..z rows from z table: ) r%rr9rHr<r=rfields_get_dataframe_results Exception)r&r?querydfZ columns_strZ sample_rowsZsample_rows_strr(r(r)rFs    zSparkSQL._get_sample_spark_rowsr tuple)r6r/cCsttt|Sr5)rUr=r+asDictvalues)r&r6r(r(r)_convert_row_as_tupleszSparkSQL._convert_row_as_tuplerr<)rTr/cCstt|j|Sr5)r<r=rXr;)r&rTr(r(r)rQszSparkSQL._get_dataframe_resultsall)commandfetchr/cCs,|j|}|dkr|d}t||S)None)rr9limitr+rQ)r&rZr[rTr(r(r)runs  z SparkSQL.runc Cs@z ||WSty:}zd|WYd}~Sd}~00dS)afGet information about specified tables. Follows best practices as specified in: Rajkumar et al, 2022 (https://arxiv.org/abs/2204.00498) If `sample_rows_in_table_info`, the specified number of sample rows will be appended to each table description. This can increase performance as demonstrated in the paper. Error: N)rIr )r&rCer(r(r)get_table_info_no_throws  z SparkSQL.get_table_info_no_throwc CsBz|||WSty<}zd|WYd}~Sd}~00dS)a*Execute a SQL command and return a string representing the results. If the statement returns rows, a string of the results is returned. If the statement returns no rows, an empty string is returned. If the statement throws an error, the error message is returned. r`N)r_rR)r&rZr[rar(r(r) run_no_throwszSparkSQL.run_no_throw)NNNNNr )N)N)rY)N)rY)__name__ __module__ __qualname____doc__r* classmethodr2r"rrBrIrFrXrQr_rbrcr(r(r(r)r s*?r N) __future__rtypingrrrrrrrr r r r(r(r(r)s