{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/23263","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/23263/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/23263/comments","events_url":"https://api.github.com/repos/apache/spark/issues/23263/events","html_url":"https://github.com/apache/spark/pull/23263","id":388905875,"node_id":"MDExOlB1bGxSZXF1ZXN0MjM3MDU4MjQ3","number":23263,"title":"[SPARK-23674][ML] Adds Spark ML Events to Instrumentation","user":{"login":"HyukjinKwon","id":6477701,"node_id":"MDQ6VXNlcjY0Nzc3MDE=","avatar_url":"https://avatars0.githubusercontent.com/u/6477701?v=4","gravatar_id":"","url":"https://api.github.com/users/HyukjinKwon","html_url":"https://github.com/HyukjinKwon","followers_url":"https://api.github.com/users/HyukjinKwon/followers","following_url":"https://api.github.com/users/HyukjinKwon/following{/other_user}","gists_url":"https://api.github.com/users/HyukjinKwon/gists{/gist_id}","starred_url":"https://api.github.com/users/HyukjinKwon/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HyukjinKwon/subscriptions","organizations_url":"https://api.github.com/users/HyukjinKwon/orgs","repos_url":"https://api.github.com/users/HyukjinKwon/repos","events_url":"https://api.github.com/users/HyukjinKwon/events{/privacy}","received_events_url":"https://api.github.com/users/HyukjinKwon/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":41,"created_at":"2018-12-08T09:43:54Z","updated_at":"2018-12-24T06:03:00Z","closed_at":null,"author_association":"MEMBER","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/23263","html_url":"https://github.com/apache/spark/pull/23263","diff_url":"https://github.com/apache/spark/pull/23263.diff","patch_url":"https://github.com/apache/spark/pull/23263.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR proposes to add ML events to Instrumentation, and use it in Pipeline so that other developers can track and add some actions for them.\r\n\r\n## Introduction\r\n\r\nML events (like SQL events) can be quite useful when people want to track and make some actions for corresponding ML operations. For instance, I have been working on integrating \r\nApache Spark with [Apache Atlas](https://atlas.apache.org/QuickStart.html). With some custom changes with this PR, I can visualise ML pipeline as below:\r\n\r\n![spark_ml_streaming_lineage](https://user-images.githubusercontent.com/6477701/49682779-394bca80-faf5-11e8-85b8-5fae28b784b3.png)\r\n\r\nAnother good thing that might have to be considered is, that we can interact this with other SQL/Streaming events. For instance, where the input `Dataset` is originated. For instance, with current Apache Spark, I can visualise SQL operations as below:\r\n\r\n![screen shot 2018-12-10 at 9 41 36 am](https://user-images.githubusercontent.com/6477701/49706269-d9bdfe00-fc5f-11e8-943a-3309d1856ba5.png)\r\n\r\nI think we can combine those existing lineages together to easily understand where the data comes and goes. Currently, ML side is a hole so the lineages can't be connected for the current Apache Spark ..\r\n\r\nTo add up, I think it's not to mention how useful it is to track the SQL/Streaming operations. Likewise, I would like to propose ML events as well (as lowest stability `@Unstable` APIs for now - no guarantee about stability).\r\n\r\n## Implementation Details\r\n\r\n### Sends event (but not expose ML specific listener)\r\n\r\n**`mllib/src/main/scala/org/apache/spark/ml/events.scala`**\r\n\r\n```scala\r\n@Unstable\r\ncase class ...StartEvent(caller, input)\r\n@Unstable\r\ncase class ...EndEvent(caller, output)\r\n\r\ntrait MLEvents {\r\n  // Wrappers to send events:\r\n  // def with...Event(body) = {\r\n  //   body()\r\n  //   SparkContext.getOrCreate().listenerBus.post(event)\r\n  // }\r\n}\r\n```\r\n\r\nThis trait is used by `Instrumentation`.\r\n\r\n```scala\r\nclass Instrumentation ... with MLEvents {\r\n```\r\n\r\nand used as below:\r\n\r\n```scala\r\ninstrumented { instr =>\r\n  instr.with...Event(...) {\r\n    ...\r\n  }\r\n}\r\n```\r\n\r\n\r\nThis way mimics both:\r\n\r\n**1. Catalog events (see `org/apache/spark/sql/catalyst/catalog/events.scala`)**\r\n\r\n- This allows a Catalog specific listener to be added `ExternalCatalogEventListener` \r\n\r\n- It's implemented in a way of wrapping whole `ExternalCatalog` named `ExternalCatalogWithListener`\r\nwhich delegates the operations to `ExternalCatalog`\r\n\r\nThis is not quite possible in this case because most of instances (like `Pipeline`) will be directly created in most of cases. We might be able to do that via extending `ListenerBus` for all possible instances but IMHO it's too invasive. Also, exposing another ML specific listener sounds a bit too much at this stage. Therefore, I simply borrowed file name and structures here\r\n\r\n**2. SQL execution events (see `org/apache/spark/sql/execution/SQLExecution.scala`)**\r\n\r\n- Add an object that wraps a body to send events\r\n\r\nCurrent apporach is rather close to this. It has a `with...` wrapper to send events. I borrowed this approach to be consistent.\r\n\r\n## Usage\r\n\r\nIt needs a custom implementation for a query listener. For instance,\r\n\r\nwith the custom listener below:\r\n\r\n```scala\r\nclass CustomMLListener extends SparkListener\r\n  def onOtherEvents(e) = e match {\r\n    case e: MLEvent => // do something\r\n    case _ => // pass\r\n  }\r\n}\r\n```\r\n\r\nThere are two (existing) ways to use this.\r\n\r\n```scala\r\nspark.sparkContext.addSparkListener(new CustomMLListener)\r\n```\r\n\r\n```bash\r\nspark-submit ...\\\r\n  --conf spark.extraListeners=CustomMLListener\\\r\n  ...\r\n```\r\n\r\nIt's also similar with other existing implementation in SQL side.\r\n\r\n## Target users\r\n\r\n1. I think someone in general would likely utilise this feature like other event listeners. At least, I can see some interests going on outside.\r\n\r\n    - SQL Listener\r\n      - https://stackoverflow.com/questions/46409339/spark-listener-to-an-sql-query\r\n      - http://apache-spark-user-list.1001560.n3.nabble.com/spark-sql-Custom-Query-Execution-listener-via-conf-properties-td30979.html\r\n\r\n    - Streaming Query Listener\r\n      - https://jhui.github.io/2017/01/15/Apache-Spark-Streaming/\r\n      -  http://apache-spark-developers-list.1001551.n3.nabble.com/Structured-Streaming-with-Watermark-td25413.html#a25416\r\n\r\n2. Someone would likely run this via Atlas. The plugin mirror intentionally is exposed at [spark-atlas-connector](https://github.com/hortonworks-spark/spark-atlas-connector) so that anyone could do something about lineage and governance in Atlas. I'm trying to show integrated lineages in Apache Spark but this is a missing hole.\r\n\r\n## How was this patch tested?\r\n\r\nManually tested and unit tests were added.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/449690201","html_url":"https://github.com/apache/spark/pull/23263#issuecomment-449690201","issue_url":"https://api.github.com/repos/apache/spark/issues/23263","id":449690201,"node_id":"MDEyOklzc3VlQ29tbWVudDQ0OTY5MDIwMQ==","user":{"login":"AmplabJenkins","id":4031291,"node_id":"MDQ6VXNlcjQwMzEyOTE=","avatar_url":"https://avatars3.githubusercontent.com/u/4031291?v=4","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2018-12-24T06:03:00Z","updated_at":"2018-12-24T06:03:00Z","author_association":"NONE","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}