o biS(@sBdZddlZddlZddlZddlmZmZddlmZm Z ddl m Z m Z m Z mZmZmZmZddlZddlmZmZmZddlmZe rRddlmZddlmZeeZd ZeGd d d Z eGd d d Z!eGdddZ"eGdddZ#ddZ$de#de fddZ%dddZ&GdddeZ'Gddde'Z(dS)z,Metadata exporter API for Ray Data datasets.N)ABCabstractmethod) dataclassfield) TYPE_CHECKINGAnyDictListMappingOptionalSequence) EventLogTypecheck_export_api_enabledget_export_event_logger) DataContext)PhysicalOperatorunknownc@s"eZdZUdZeed<eed<dS)SubStagezRepresents a sub-stage within an operator in the DAG. Attributes: name: The name of the sub-stage. id: The unique identifier of the sub-stage. nameidN)__name__ __module__ __qualname____doc__str__annotations__rrX/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/metadata_exporter.pyrs  rc@speZdZUdZeed<eed<eed<eedZe eed<eedZ e e ed<ee dZ eeefed<d S) OperatoraRepresents a data processing operator in the DAG. Attributes: name: The name of the operator. id: The unique identifier of the operator within the DAG structure, typically incorporating a position or index (e.g., "ReadParquet_0"). This is used for referencing operators within the DAG topology. uuid: The system-generated UUID of the physical operator instance. This is the internal unique identifier created when the operator instance is initialized and remains consistent throughout its lifetime. input_dependencies: List of operator IDs that this operator depends on for input. sub_stages: List of sub-stages contained within this operator. args: User-specified arguments associated with the operator, which may include configuration settings, options, or other relevant data for the operator. rruuiddefault_factoryinput_dependencies sub_stagesargsN)rrrrrrrlistr"r r#rdictr$rrrrrrr+s rc@sJeZdZUdZeedZeee d<e ddde de fddfdd Z d S) TopologyzRepresents the complete structure of the operator DAG. Attributes: operators: List of all operators in the DAG. r operatorsdagrop_to_idreturnc st}|D]F}|}t|j||jfdd|jDt|d}t|drG|j rGt |j D]\}}|d|}|j t ||dq1|j |q|S)zCreate a Topology structure from the physical operator DAG. Args: dag: The operator DAG to analyze. Returns: A Topology object representing the operator DAG structure. csg|] }|vr|qSrr).0depr*rr gsz5Topology.create_topology_metadata..)rrrr"r$_sub_progress_bar_names_sub_rr)r'post_order_iterrrrr"sanitize_for_struct_get_logical_argshasattrr0 enumerater#appendrr() r)r*resultopop_idoperatorjsub_name sub_stage_idrr.rcreate_topology_metadataOs$    z!Topology.create_topology_metadataN)rrrrrr%r(r rr staticmethodrrr@rrrrr'Es  r'c@s:eZdZUdZeed<eed<eed<eed<eed<dS)DatasetMetadataaMetadata about a Ray Data dataset. This class represents the metadata associated with a dataset, including its provenance information and execution details. Attributes: job_id: The ID of the job running this dataset. topology: The structure of the dataset's operator DAG. dataset_id: The unique ID of the dataset. start_time: The timestamp when the dataset execution started. data_context: The DataContext attached to the dataset. job_idtopology dataset_id start_time data_contextN) rrrrrrr'floatrrrrrrBws   rBc Cst|trdd|DSt|ttttfs|dur|St|tr)dd|DSzt |WSt t fyMzt|WYSt yLt YYSww)NcSsi|] \}}|t|qSrr4)r,kvrrr sz'sanitize_for_struct..cSsg|]}t|qSrrI)r,rKrrrr/sz'sanitize_for_struct..) isinstancer itemsrintrHboolr jsondumps TypeError OverflowError ExceptionUNKNOWN)objrrrr4s     r4dataset_metadatar+cCsddlm}ddlm}ddlm}m}m}m}|}|}|j j D]<} |} | | j || j | j| j| d} | jD]} | j| q;| jD]} || j | jd}| j|qG|j | q"|}| t||j||j|j|j|d}|j ||S)aConvert the dataset metadata to a protobuf message. Args: dataset_metadata: DatasetMetadata object containing the dataset's information and DAG structure. Returns: The protobuf message representing the dataset metadata. r)asdict)Struct)ExportDatasetMetadatarrr')rrrr$r2)rErCrFrG) dataclassesrYgoogle.protobuf.struct_pb2rZ.ray.core.generated.export_dataset_metadata_pb2r[rrr'rDr(updater$rrrr"r8r#r4rGrErCrFCopyFrom)rXrYrZProtoDatasetMetadata ProtoOperator ProtoSubStage ProtoTopologyproto_dataset_metadataproto_topologyr:r$proto_operatordep_id sub_stageproto_sub_stagerGrrrdataset_metadata_to_protosB      rkDatasetMetadataExportercCstS)znGet the dataset metadata exporter instance. Returns: The dataset metadata exporter instance. )LoggerDatasetMetadataExportercreate_if_enabledrrrrget_dataset_metadata_exportersroc@s@eZdZdZededdfddZeededfddZ dS) rlzAbstract base class for dataset metadata exporters. Implementations of this interface can export Ray Data metadata to various destinations like log files, databases, or monitoring systems. rXr+NcCdS)zExport dataset metadata to the destination. Args: dataset_metadata: DatasetMetadata object containing dataset information. Nr)selfrXrrrexport_dataset_metadatasz/DatasetMetadataExporter.export_dataset_metadatacCrp)zCreate an exporter instance if the export functionality is enabled. Returns: An exporter instance if enabled, None otherwise. Nr)clsrrrrnsz)DatasetMetadataExporter.create_if_enabled) rrrrrrBrr classmethodr rnrrrrrlsc@sHeZdZdZdejfddZdeddfdd Ze de dfd d Z dS) rmzDataset metadata exporter implementation that uses the Ray export event logger. This exporter writes dataset metadata to log files using Ray's export event system. loggercCs ||_dS)zInitialize with a configured export event logger. Args: logger: The export event logger to use for writing events. N)_export_logger)rqrurrr__init__ s z&LoggerDatasetMetadataExporter.__init__rXr+NcCst|}|j|dS)zExport dataset metadata using the export event logger. Args: dataset_metadata: DatasetMetadata object containing dataset information. N)rkrv send_event)rqrXdata_metadata_protorrrrrsz5LoggerDatasetMetadataExporter.export_dataset_metadatacCsnddlm}t|jj}|sdStjtj j j d}z t tj|}t|WSty6|dYdSw)zCreate a logger-based exporter if the export API is enabled. Returns: A LoggerDatasetMetadataExporter instance if enabled, None otherwise. r) ExportEventNlogszcUnable to initialize the export event logger, so no Dataset Metadata export events will be written.)#ray.core.generated.export_event_pb2rzr SourceTypeEXPORT_DATASET_METADATAospathjoinray_privateworker _global_nodeget_session_dir_pathrr DATASET_METADATArmrU exception)rsrz&is_dataset_metadata_export_api_enabled log_directoryrurrrrns*   z/LoggerDatasetMetadataExporter.create_if_enabled) rrrrloggingLoggerrwrBrrrtr rnrrrrrms  rm)r+rl))rrQrrabcrrr\rrtypingrrrr r r r r&ray._private.event.export_event_loggerr rrray.data.contextrray.data9ray.data._internal.execution.interfaces.physical_operatorr getLoggerrrurVrrr'rBr4rkrorlrmrrrrs8$     1 A