U h`@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}Wntk r,tdYnX|r6|n|j|_|dk rX|jj||dk rn|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(K/tmp/pip-unpacked-wheel-9gdii04g/langchain_community/utilities/spark_sql.py__init__ s@      zSparkSQL.__init__strzOptional[dict]r) database_uri engine_argskwargsreturncKsJzddlm}Wntk r,tdYnX|j|}||f|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&Zrowsr(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|}|dk r6t||}|r2td|d|}g}|D]D}||}|jrx|d7}|d||d7}|d7}||q>d|}|S)Nz table_names rz /* z*/z ) r"r differencer rAr%_get_sample_spark_rowsappendjoin)r&rBZall_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 k rtd }YnX|jd |d |d|S) NzSELECT * FROM z LIMIT  cSs|jSr5)name)fr(r(r)r7r8z1SparkSQL._get_sample_spark_rows..rCcSsg|]}d|qS)rI)rG).0r6r(r(r) sz3SparkSQL._get_sample_spark_rows..z rows from z table: ) r%rr9rGr<r=rfields_get_dataframe_results Exception)r&r>querydfZ columns_strZ sample_rowsZsample_rows_strr(r(r)rEs   zSparkSQL._get_sample_spark_rowsr tuple)r6r/cCsttt|Sr5)rTr=r+asDictvalues)r&r6r(r(r)_convert_row_as_tupleszSparkSQL._convert_row_as_tuplerr<)rSr/cCstt|j|Sr5)r<r=rWr;)r&rSr(r(r)rPszSparkSQL._get_dataframe_resultsall)commandfetchr/cCs,|j|}|dkr|d}t||S)NZone)rr9limitr+rP)r&rYrZrSr(r(r)runs  z SparkSQL.runc CsBz ||WStk r<}zd|WYSd}~XYnXdS)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)rHr )r&rBer(r(r)get_table_info_no_throws  z SparkSQL.get_table_info_no_throwc CsDz|||WStk r>}zd|WYSd}~XYnXdS)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]rQ)r&rYrZr_r(r(r) run_no_throwszSparkSQL.run_no_throw)NNNNNr )N)N)rX)N)rX)__name__ __module__ __qualname____doc__r* classmethodr2r"rrArHrErWrPr]r`rar(r(r(r)r s*?r N) __future__rtypingrrrrrrrr r r r(r(r(r)s