a bg< @s~ddlZddlZddlZddlmZmZmZmZmZm Z ddl m Z ddl m Z eeZerjddlmZGddde ZdS)N) TYPE_CHECKINGAnyIteratorListOptionalTuple)Document) BaseLoader) SparkSessionc@sheZdZdZdedeeeedddZe e e fd d d Z e e d d d Zee d ddZdS)PySparkDataFrameLoaderzLoad `PySpark` DataFrames.Ntext皙?r ) spark_sessiondfpage_content_columnfraction_of_memorycCszddlm}m}Wnty.tdYn0|r8|n|j|_t||s`tdt |||_ ||_ ||_ | \|_|_|j jt|_|j j|_dS)agInitialize with a Spark DataFrame object. Args: spark_session: The SparkSession object. df: The Spark DataFrame object. page_content_column: The name of the column containing the page content. Defaults to "text". fraction_of_memory: The fraction of memory to use. Defaults to 0.1. r) DataFramer zFpyspark is not installed. Please install it with `pip install pyspark`z3Expected data_frame to be a PySpark DataFrame, got N) pyspark.sqlrr ImportErrorZbuilderZ getOrCreateZspark isinstance ValueErrortyperrr get_num_rowsnum_rows max_num_rowsZrddmaplistrdd_dfcolumns column_names)selfrrrrrr r!/var/www/html/cobodadashboardai.evdpl.com/venv/lib/python3.9/site-packages/langchain_community/document_loaders/pyspark_dataframe.py__init__s$    zPySparkDataFrameLoader.__init__)returnc Csz ddl}Wn.ty:}ztd|WYd}~n d}~00|jdd}t|}|}|j}t |||j }t ||j |fS)z4Gets the number of "feasible" rows for the DataFramerNzBpsutil not installed. Please install it with `pip install psutil`.) psutilrrlimitZcollectsys getsizeofZvirtual_memory availableintrmincount)r r&erowZestimated_row_sizeZmem_infoZavailable_memoryrr!r!r"r:s    z#PySparkDataFrameLoader.get_num_rowsc#sTjD]DfddttD}|j}|jt||dVq dS)z#A lazy loader for document content.csi|]}j||qSr!)r).0ir/r r!r" Nz4PySparkDataFrameLoader.lazy_load..)Z page_contentmetadataN)rZtoLocalIteratorrangelenrpopr)r r5r r!r2r" lazy_loadKs   z PySparkDataFrameLoader.lazy_loadcCsJ|j|jkr0td|jd|jd|}tt ||jS)zLoad from the dataframe.z The number of DataFrame rows is zQ, but we will only include the amount of rows that can reasonably fit in memory: .) rr-rloggerwarningrr9r itertoolsislice)r Zlazy_load_iteratorr!r!r"loadSszPySparkDataFrameLoader.load)NNr r )__name__ __module__ __qualname____doc__rrstrfloatr#rr+rrrr9rr?r!r!r!r"r s 'r )r=loggingr(typingrrrrrrZlangchain_core.documentsrZ)langchain_community.document_loaders.baser getLogger__file__r;rr r r!r!r!r"s