o ॵi@s"ddlZddlZGdddZdS)Nc @seZdZdZddZddZdededejfd d Z deded edd fd dZ e ddZ e de de dededef ddZe de de de dededef ddZddedefddZd S)MaxComputeUtila MaxCompute util class. Args: access_id: your access id of MaxCompute access_key: access key of MaxCompute project_name: your project name of MaxCompute endpoint: endpoint of MaxCompute Attributes: _odps: ODPS object cCs ddlm}||||||_dS)Nr)ODPS)odpsr_odps)self access_id access_key project_nameendpointrr `/home/ubuntu/.local/lib/python3.10/site-packages/modelscope/msdatasets/utils/maxcompute_utils.py__init__s zMaxComputeUtil.__init__cCs |j|S)z. Get MaxCompute table object. )r get_table)r table_namer r r _get_tables zMaxComputeUtil._get_tabler pt_conditionreturncCsF||}|j|dd }|}Wd|S1swY|S)z Read data from MaxCompute table. :param table_name: table name :param pt_condition: partition condition, Example: pt_condition = 'dt=20230331' :return: pandas dataframe with all data F) partitionlimitN)r open_reader to_pandas)rrrtreaderpd_dfr r r _read_data!s   zMaxComputeUtil._read_data output_pathNcCs.|||}|j|ddtd|ddS)a  Fetch data from MaxCompute table to local file. :param table_name: table name :param pt_condition: partition condition, Example: pt_condition = 'dt=20230331' :param output_path: output path :return: None F)indexzFetch data to z successfully.N)rto_csvprint)rrrrrr r r fetch_data_to_csv0s z MaxComputeUtil.fetch_data_to_csvcCs4|s|j}|dkrtd|||kr|}||fS)Nrz%batch_size must be positive, but got )count ValueError)r batch_sizerr r r _check_batch_args>sz MaxComputeUtil._check_batch_args batch_size_inlimit_in drop_last_in partitionscolumnsc cst|||\}}t||}t|dD]R}||kr4|s4||dkr4|||||||}n ||||d|}g} |D]} ddt| D} | dt| t|} | | qDtj | |dVqdS)a Generate batch data from MaxCompute table. Args: reader: MaxCompute table reader batch_size_in: batch size limit_in: limit of data, None means fetch all data drop_last_in: whether drop last incomplete batch data partitions: table partitions columns: table columns Returns: batch data generator rcSg|]\}}|qSr r .0_valr r r hz3MaxComputeUtil.gen_reader_batch..Nr() rr#mathfloorrangelistlenappendpd DataFrame) rr$r%r&r'r( batch_numi batch_recordsbatch_data_listrecordtmp_valsr r r gen_reader_batchIs( zMaxComputeUtil.gen_reader_batchrcCst|||\}}|rt||}nt||}|dkr%td|||kr3td|d|||}|d|} | |krC|} ||| } g} | D]} ddt| D} | dt| t|} | | qMt j | |d S) a Get single batch data from MaxCompute table by indexing. Args: reader: MaxCompute table reader index: index of batch data batch_size_in: batch size limit_in: limit of data, None means fetch all data drop_last_in: whether drop last incomplete batch data partitions: table partitions columns: table columns Returns: single batch data (dataframe) rz$index must be non-negative, but got z1index must be less than batch_num, but got index=z , batch_num=r)cSr*r r r+r r r r/r0z2MaxComputeUtil.gen_reader_item..Nr1) rr#r2r3ceilr!r5r6r7r8r9)rrr$r%r&r'r(r:startend batch_itemr=r>r?r r r gen_reader_itemms.   zMaxComputeUtil.gen_reader_itemcCsB||}|j|d }||fWdS1swYdS)N)r)rr)rrr table_insrr r r get_table_reader_inss $z#MaxComputeUtil.get_table_reader_ins)N)__name__ __module__ __qualname____doc__r rstrr8r9rr staticmethodr#intboolr5r@rErGr r r r rs:     # .r)r2pandasr8rr r r r s