{"action":"created","comment":{"url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/comments/24186099","id":24186099,"diff_hunk":"@@ -0,0 +1,80 @@\n+package com.datastax.spark.connector.metrics\n+\n+import com.codahale.metrics.Timer\n+import com.datastax.driver.core.Row\n+import org.apache.spark.{SparkEnv, TaskContext}\n+import org.apache.spark.executor.{DataReadMethod, InputMetrics}\n+import org.apache.spark.metrics.CassandraConnectorSource\n+\n+private[connector] trait InputMetricsUpdater {\n+  def resultSetFetchTimer: Option[Timer]\n+\n+  def updateMetrics(row: Row): Row\n+\n+  def finish(): Long\n+}\n+\n+private class DetailedInputMetricsUpdater(metrics: InputMetrics, groupSize: Int) extends InputMetricsUpdater {\n+  require(groupSize > 0)\n+\n+  val resultSetFetchTimer = Some(CassandraConnectorSource.readMoreRowsTimer)\n+\n+  private val taskTimer = CassandraConnectorSource.readTaskTimer.time()\n+\n+  private var cnt = 0\n+  private var dataLength = metrics.bytesRead\n+\n+  def updateMetrics(row: Row): Row = {\n+    for (i <- 0 until row.getColumnDefinitions.size() if !row.isNull(i))\n+      dataLength += row.getBytesUnsafe(i).remaining()\n+\n+    cnt += 1\n+    if (cnt == groupSize)\n+      update()\n+    row\n+  }\n+\n+  @inline\n+  private def update(): Unit = {\n+    CassandraConnectorSource.readRowMeter.mark(cnt)\n+    CassandraConnectorSource.readByteMeter.mark(metrics.bytesRead - dataLength)\n+    dataLength = metrics.bytesRead\n+    cnt = 0\n+  }\n+\n+  def finish(): Long = {\n+    update()\n+    taskTimer.stop()\n+  }\n+}\n+\n+private class DummyInputMetricsUpdater extends InputMetricsUpdater {\n+  private val taskTimer = System.nanoTime()\n+\n+  val resultSetFetchTimer = None\n+\n+  def updateMetrics(row: Row): Row = row\n+\n+  def finish(): Long = {\n+    System.nanoTime() - taskTimer\n+  }\n+}\n+\n+object InputMetricsUpdater {\n+  lazy val detailedMetricsEnabled =\n+    SparkEnv.get.conf.getBoolean(\"spark.cassandra.input.metrics\", defaultValue = true)\n+\n+  def apply(taskContext: TaskContext, groupSize: Int): InputMetricsUpdater = {\n+    CassandraConnectorSource.ensureInitialized\n+\n+    if (detailedMetricsEnabled) {\n+      val tm = taskContext.taskMetrics()\n+      if (tm.inputMetrics.isEmpty || tm.inputMetrics.get.readMethod != DataReadMethod.Hadoop)","path":"spark-cassandra-connector/src/main/scala/com/datastax/spark/connector/metrics/InputMetricsUpdater.scala","position":72,"original_position":72,"commit_id":"8776f01b737bd538af89b7b8fac1bb751b8b1d13","original_commit_id":"8776f01b737bd538af89b7b8fac1bb751b8b1d13","user":{"login":"RussellSpitzer","id":413025,"avatar_url":"https://avatars.githubusercontent.com/u/413025?v=3","gravatar_id":"","url":"https://api.github.com/users/RussellSpitzer","html_url":"https://github.com/RussellSpitzer","followers_url":"https://api.github.com/users/RussellSpitzer/followers","following_url":"https://api.github.com/users/RussellSpitzer/following{/other_user}","gists_url":"https://api.github.com/users/RussellSpitzer/gists{/gist_id}","starred_url":"https://api.github.com/users/RussellSpitzer/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/RussellSpitzer/subscriptions","organizations_url":"https://api.github.com/users/RussellSpitzer/orgs","repos_url":"https://api.github.com/users/RussellSpitzer/repos","events_url":"https://api.github.com/users/RussellSpitzer/events{/privacy}","received_events_url":"https://api.github.com/users/RussellSpitzer/received_events","type":"User","site_admin":false},"body":"Same comment with the DataWriteMethod. I'd rather not if at any point we make any statement containing \"hadoop\" to end users","created_at":"2015-02-05T18:31:19Z","updated_at":"2015-02-05T18:31:19Z","html_url":"https://github.com/datastax/spark-cassandra-connector/pull/536#discussion_r24186099","pull_request_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536","_links":{"self":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/comments/24186099"},"html":{"href":"https://github.com/datastax/spark-cassandra-connector/pull/536#discussion_r24186099"},"pull_request":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536"}}},"pull_request":{"url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536","id":28632590,"html_url":"https://github.com/datastax/spark-cassandra-connector/pull/536","diff_url":"https://github.com/datastax/spark-cassandra-connector/pull/536.diff","patch_url":"https://github.com/datastax/spark-cassandra-connector/pull/536.patch","issue_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/536","number":536,"state":"open","locked":false,"title":"SPARKC-27: Refactored the read path and the write path in order to provide read/write metrics for the Spark metrics system","user":{"login":"jacek-lewandowski","id":6516951,"avatar_url":"https://avatars.githubusercontent.com/u/6516951?v=3","gravatar_id":"","url":"https://api.github.com/users/jacek-lewandowski","html_url":"https://github.com/jacek-lewandowski","followers_url":"https://api.github.com/users/jacek-lewandowski/followers","following_url":"https://api.github.com/users/jacek-lewandowski/following{/other_user}","gists_url":"https://api.github.com/users/jacek-lewandowski/gists{/gist_id}","starred_url":"https://api.github.com/users/jacek-lewandowski/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jacek-lewandowski/subscriptions","organizations_url":"https://api.github.com/users/jacek-lewandowski/orgs","repos_url":"https://api.github.com/users/jacek-lewandowski/repos","events_url":"https://api.github.com/users/jacek-lewandowski/events{/privacy}","received_events_url":"https://api.github.com/users/jacek-lewandowski/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-02-04T11:52:07Z","updated_at":"2015-02-05T18:31:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"f79ac8065df290ba4d64cb0e4a4ec508e00e8460","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536/commits","review_comments_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536/comments","review_comment_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/comments/{number}","comments_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/536/comments","statuses_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/statuses/8776f01b737bd538af89b7b8fac1bb751b8b1d13","head":{"label":"datastax:SPARKC-27","ref":"SPARKC-27","sha":"8776f01b737bd538af89b7b8fac1bb751b8b1d13","user":{"login":"datastax","id":573369,"avatar_url":"https://avatars.githubusercontent.com/u/573369?v=3","gravatar_id":"","url":"https://api.github.com/users/datastax","html_url":"https://github.com/datastax","followers_url":"https://api.github.com/users/datastax/followers","following_url":"https://api.github.com/users/datastax/following{/other_user}","gists_url":"https://api.github.com/users/datastax/gists{/gist_id}","starred_url":"https://api.github.com/users/datastax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/datastax/subscriptions","organizations_url":"https://api.github.com/users/datastax/orgs","repos_url":"https://api.github.com/users/datastax/repos","events_url":"https://api.github.com/users/datastax/events{/privacy}","received_events_url":"https://api.github.com/users/datastax/received_events","type":"Organization","site_admin":false},"repo":{"id":21280887,"name":"spark-cassandra-connector","full_name":"datastax/spark-cassandra-connector","owner":{"login":"datastax","id":573369,"avatar_url":"https://avatars.githubusercontent.com/u/573369?v=3","gravatar_id":"","url":"https://api.github.com/users/datastax","html_url":"https://github.com/datastax","followers_url":"https://api.github.com/users/datastax/followers","following_url":"https://api.github.com/users/datastax/following{/other_user}","gists_url":"https://api.github.com/users/datastax/gists{/gist_id}","starred_url":"https://api.github.com/users/datastax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/datastax/subscriptions","organizations_url":"https://api.github.com/users/datastax/orgs","repos_url":"https://api.github.com/users/datastax/repos","events_url":"https://api.github.com/users/datastax/events{/privacy}","received_events_url":"https://api.github.com/users/datastax/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/datastax/spark-cassandra-connector","description":"If you write a Spark application that needs access to Cassandra, this library is for you","fork":false,"url":"https://api.github.com/repos/datastax/spark-cassandra-connector","forks_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/forks","keys_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/keys{/key_id}","collaborators_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/teams","hooks_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/hooks","issue_events_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/events{/number}","events_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/events","assignees_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/assignees{/user}","branches_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/branches{/branch}","tags_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/tags","blobs_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/refs{/sha}","trees_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/trees{/sha}","statuses_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/statuses/{sha}","languages_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/languages","stargazers_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/stargazers","contributors_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/contributors","subscribers_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/subscribers","subscription_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/subscription","commits_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/commits{/sha}","git_commits_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/commits{/sha}","comments_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/comments{/number}","issue_comment_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/comments/{number}","contents_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/contents/{+path}","compare_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/compare/{base}...{head}","merges_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/merges","archive_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/downloads","issues_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues{/number}","pulls_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls{/number}","milestones_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/milestones{/number}","notifications_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/labels{/name}","releases_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/releases{/id}","created_at":"2014-06-27T15:45:54Z","updated_at":"2015-02-05T15:50:20Z","pushed_at":"2015-02-05T15:59:10Z","git_url":"git://github.com/datastax/spark-cassandra-connector.git","ssh_url":"git@github.com:datastax/spark-cassandra-connector.git","clone_url":"https://github.com/datastax/spark-cassandra-connector.git","svn_url":"https://github.com/datastax/spark-cassandra-connector","homepage":"","size":5838,"stargazers_count":370,"watchers_count":370,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":106,"mirror_url":null,"open_issues_count":39,"forks":106,"open_issues":39,"watchers":370,"default_branch":"master"}},"base":{"label":"datastax:master","ref":"master","sha":"6513b9afce3beacae5d89180977a224226e9af45","user":{"login":"datastax","id":573369,"avatar_url":"https://avatars.githubusercontent.com/u/573369?v=3","gravatar_id":"","url":"https://api.github.com/users/datastax","html_url":"https://github.com/datastax","followers_url":"https://api.github.com/users/datastax/followers","following_url":"https://api.github.com/users/datastax/following{/other_user}","gists_url":"https://api.github.com/users/datastax/gists{/gist_id}","starred_url":"https://api.github.com/users/datastax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/datastax/subscriptions","organizations_url":"https://api.github.com/users/datastax/orgs","repos_url":"https://api.github.com/users/datastax/repos","events_url":"https://api.github.com/users/datastax/events{/privacy}","received_events_url":"https://api.github.com/users/datastax/received_events","type":"Organization","site_admin":false},"repo":{"id":21280887,"name":"spark-cassandra-connector","full_name":"datastax/spark-cassandra-connector","owner":{"login":"datastax","id":573369,"avatar_url":"https://avatars.githubusercontent.com/u/573369?v=3","gravatar_id":"","url":"https://api.github.com/users/datastax","html_url":"https://github.com/datastax","followers_url":"https://api.github.com/users/datastax/followers","following_url":"https://api.github.com/users/datastax/following{/other_user}","gists_url":"https://api.github.com/users/datastax/gists{/gist_id}","starred_url":"https://api.github.com/users/datastax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/datastax/subscriptions","organizations_url":"https://api.github.com/users/datastax/orgs","repos_url":"https://api.github.com/users/datastax/repos","events_url":"https://api.github.com/users/datastax/events{/privacy}","received_events_url":"https://api.github.com/users/datastax/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/datastax/spark-cassandra-connector","description":"If you write a Spark application that needs access to Cassandra, this library is for you","fork":false,"url":"https://api.github.com/repos/datastax/spark-cassandra-connector","forks_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/forks","keys_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/keys{/key_id}","collaborators_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/teams","hooks_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/hooks","issue_events_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/events{/number}","events_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/events","assignees_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/assignees{/user}","branches_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/branches{/branch}","tags_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/tags","blobs_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/refs{/sha}","trees_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/trees{/sha}","statuses_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/statuses/{sha}","languages_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/languages","stargazers_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/stargazers","contributors_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/contributors","subscribers_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/subscribers","subscription_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/subscription","commits_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/commits{/sha}","git_commits_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/git/commits{/sha}","comments_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/comments{/number}","issue_comment_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/comments/{number}","contents_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/contents/{+path}","compare_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/compare/{base}...{head}","merges_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/merges","archive_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/downloads","issues_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues{/number}","pulls_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls{/number}","milestones_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/milestones{/number}","notifications_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/labels{/name}","releases_url":"https://api.github.com/repos/datastax/spark-cassandra-connector/releases{/id}","created_at":"2014-06-27T15:45:54Z","updated_at":"2015-02-05T15:50:20Z","pushed_at":"2015-02-05T15:59:10Z","git_url":"git://github.com/datastax/spark-cassandra-connector.git","ssh_url":"git@github.com:datastax/spark-cassandra-connector.git","clone_url":"https://github.com/datastax/spark-cassandra-connector.git","svn_url":"https://github.com/datastax/spark-cassandra-connector","homepage":"","size":5838,"stargazers_count":370,"watchers_count":370,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":106,"mirror_url":null,"open_issues_count":39,"forks":106,"open_issues":39,"watchers":370,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536"},"html":{"href":"https://github.com/datastax/spark-cassandra-connector/pull/536"},"issue":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/536"},"comments":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/issues/536/comments"},"review_comments":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536/comments"},"review_comment":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/pulls/536/commits"},"statuses":{"href":"https://api.github.com/repos/datastax/spark-cassandra-connector/statuses/8776f01b737bd538af89b7b8fac1bb751b8b1d13"}}}}