{"action":"created","comment":{"url":"https://api.github.com/repos/Azure/mmlspark/pulls/comments/140891731","pull_request_review_id":65028670,"id":140891731,"diff_hunk":"@@ -0,0 +1,215 @@\n+// Copyright (C) Microsoft Corporation. All rights reserved.\n+// Licensed under the MIT License. See LICENSE in project root for information.\n+\n+package org.apache.spark.binary\n+\n+import java.io.{Closeable, InputStream}\n+import java.net.URI\n+\n+import com.microsoft.ml.spark.StreamUtilities.ZipIterator\n+import com.microsoft.ml.spark.schema.BinarySchema\n+import org.apache.commons.io.{FilenameUtils, IOUtils}\n+import org.apache.hadoop.conf.Configuration\n+import org.apache.hadoop.fs.{FileStatus, Path}\n+import org.apache.hadoop.io.BytesWritable\n+import org.apache.hadoop.mapreduce._\n+import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}\n+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl\n+import org.apache.log4j.Logger\n+import org.apache.spark.TaskContext\n+import org.apache.spark.sql.SparkSession\n+import org.apache.spark.sql.catalyst.InternalRow\n+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow\n+import org.apache.spark.sql.execution.datasources._\n+import org.apache.spark.sql.sources._\n+import org.apache.spark.sql.types._\n+import org.apache.spark.unsafe.types.UTF8String\n+import org.apache.spark.util.SerializableConfiguration\n+\n+import scala.util.Random\n+\n+/**\n+  * Actually reads the records from files\n+  *\n+  * @param subsample  what ratio to subsample\n+  * @param inspectZip whether to inspect zip files\n+  */\n+private[spark] class BinaryRecordReader(val subsample: Double, val inspectZip: Boolean)\n+  extends RecordReader[String, BytesWritable] {\n+\n+  private var done: Boolean = false\n+  private var inputStream: InputStream = _\n+  private var filename: String = _\n+  private var recordValue: BytesWritable = _\n+  private var progress: Float = 0.0F\n+  private val rng: Random = new Random()\n+  private var zipIterator: ZipIterator = _\n+\n+  override def close(): Unit = {\n+    if (inputStream != null) {\n+      inputStream.close()\n+    }\n+  }\n+\n+  override def getCurrentKey: String = {\n+    filename\n+  }\n+\n+  override def getCurrentValue: BytesWritable = {\n+    recordValue\n+  }\n+\n+  override def getProgress: Float = {\n+    progress\n+  }\n+\n+  override def initialize(inputSplit: InputSplit, context: TaskAttemptContext): Unit = {\n+    // the file input\n+    val fileSplit = inputSplit.asInstanceOf[FileSplit]\n+\n+    // the actual file we will be reading from\n+    val file = fileSplit.getPath\n+    // job configuration\n+    val conf = context.getConfiguration\n+    // get the filesystem\n+    val fs = file.getFileSystem(conf)\n+    // open the File\n+    filename = file.toString\n+\n+    inputStream = fs.open(file)\n+    rng.setSeed(filename.hashCode.toLong)\n+    if (inspectZip && FilenameUtils.getExtension(filename) == \"zip\") {\n+        zipIterator = new ZipIterator(inputStream, filename, subsample, rng)\n+    }\n+  }\n+\n+  def markAsDone(): Unit = {\n+    done = true\n+    progress = 1.0F\n+  }\n+\n+  override def nextKeyValue(): Boolean = {\n+    if (done) {\n+      return false\n+    }\n+\n+    if (zipIterator != null) {\n+      if (zipIterator.hasNext) {\n+        val (fn, barr) = zipIterator.next\n+        filename = fn\n+        recordValue = new BytesWritable()\n+        recordValue.set(barr, 0, barr.length)\n+        true\n+      } else {\n+        markAsDone()\n+        false\n+      }\n+    } else {\n+      rng.setSeed(filename.hashCode.toLong)","path":"src/readers/src/main/scala/BinaryFileFormat.scala","position":108,"original_position":108,"commit_id":"ad159a03d270a48a65bf0828e56cd4c3c23b33c6","original_commit_id":"ad159a03d270a48a65bf0828e56cd4c3c23b33c6","user":{"login":"elibarzilay","id":185905,"avatar_url":"https://avatars2.githubusercontent.com/u/185905?v=4","gravatar_id":"","url":"https://api.github.com/users/elibarzilay","html_url":"https://github.com/elibarzilay","followers_url":"https://api.github.com/users/elibarzilay/followers","following_url":"https://api.github.com/users/elibarzilay/following{/other_user}","gists_url":"https://api.github.com/users/elibarzilay/gists{/gist_id}","starred_url":"https://api.github.com/users/elibarzilay/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elibarzilay/subscriptions","organizations_url":"https://api.github.com/users/elibarzilay/orgs","repos_url":"https://api.github.com/users/elibarzilay/repos","events_url":"https://api.github.com/users/elibarzilay/events{/privacy}","received_events_url":"https://api.github.com/users/elibarzilay/received_events","type":"User","site_admin":false},"body":"A more real problem with this is that it sets the seed to the hash, which kills the randomality of any following random calls, no?","created_at":"2017-09-25T20:40:35Z","updated_at":"2017-09-25T20:40:36Z","html_url":"https://github.com/Azure/mmlspark/pull/134#discussion_r140891731","pull_request_url":"https://api.github.com/repos/Azure/mmlspark/pulls/134","author_association":"COLLABORATOR","_links":{"self":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/comments/140891731"},"html":{"href":"https://github.com/Azure/mmlspark/pull/134#discussion_r140891731"},"pull_request":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/134"}},"in_reply_to_id":140840106},"pull_request":{"url":"https://api.github.com/repos/Azure/mmlspark/pulls/134","id":141673227,"html_url":"https://github.com/Azure/mmlspark/pull/134","diff_url":"https://github.com/Azure/mmlspark/pull/134.diff","patch_url":"https://github.com/Azure/mmlspark/pull/134.patch","issue_url":"https://api.github.com/repos/Azure/mmlspark/issues/134","number":134,"state":"open","locked":false,"title":"Refactor MMLSpark for Structured Streaming","user":{"login":"mhamilton723","id":6456637,"avatar_url":"https://avatars2.githubusercontent.com/u/6456637?v=4","gravatar_id":"","url":"https://api.github.com/users/mhamilton723","html_url":"https://github.com/mhamilton723","followers_url":"https://api.github.com/users/mhamilton723/followers","following_url":"https://api.github.com/users/mhamilton723/following{/other_user}","gists_url":"https://api.github.com/users/mhamilton723/gists{/gist_id}","starred_url":"https://api.github.com/users/mhamilton723/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhamilton723/subscriptions","organizations_url":"https://api.github.com/users/mhamilton723/orgs","repos_url":"https://api.github.com/users/mhamilton723/repos","events_url":"https://api.github.com/users/mhamilton723/events{/privacy}","received_events_url":"https://api.github.com/users/mhamilton723/received_events","type":"User","site_admin":false},"body":"- Replace old readers with new performant Dataframe readers\r\n- Remove all references to DF.rdd.mapPartitions","created_at":"2017-09-18T18:34:56Z","updated_at":"2017-09-25T20:40:36Z","closed_at":null,"merged_at":null,"merge_commit_sha":"01d373cc9b9b7870d225326b4b5bb396ae9ef303","assignee":{"login":"mhamilton723","id":6456637,"avatar_url":"https://avatars2.githubusercontent.com/u/6456637?v=4","gravatar_id":"","url":"https://api.github.com/users/mhamilton723","html_url":"https://github.com/mhamilton723","followers_url":"https://api.github.com/users/mhamilton723/followers","following_url":"https://api.github.com/users/mhamilton723/following{/other_user}","gists_url":"https://api.github.com/users/mhamilton723/gists{/gist_id}","starred_url":"https://api.github.com/users/mhamilton723/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhamilton723/subscriptions","organizations_url":"https://api.github.com/users/mhamilton723/orgs","repos_url":"https://api.github.com/users/mhamilton723/repos","events_url":"https://api.github.com/users/mhamilton723/events{/privacy}","received_events_url":"https://api.github.com/users/mhamilton723/received_events","type":"User","site_admin":false},"assignees":[{"login":"mhamilton723","id":6456637,"avatar_url":"https://avatars2.githubusercontent.com/u/6456637?v=4","gravatar_id":"","url":"https://api.github.com/users/mhamilton723","html_url":"https://github.com/mhamilton723","followers_url":"https://api.github.com/users/mhamilton723/followers","following_url":"https://api.github.com/users/mhamilton723/following{/other_user}","gists_url":"https://api.github.com/users/mhamilton723/gists{/gist_id}","starred_url":"https://api.github.com/users/mhamilton723/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhamilton723/subscriptions","organizations_url":"https://api.github.com/users/mhamilton723/orgs","repos_url":"https://api.github.com/users/mhamilton723/repos","events_url":"https://api.github.com/users/mhamilton723/events{/privacy}","received_events_url":"https://api.github.com/users/mhamilton723/received_events","type":"User","site_admin":false}],"requested_reviewers":[],"milestone":null,"commits_url":"https://api.github.com/repos/Azure/mmlspark/pulls/134/commits","review_comments_url":"https://api.github.com/repos/Azure/mmlspark/pulls/134/comments","review_comment_url":"https://api.github.com/repos/Azure/mmlspark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/Azure/mmlspark/issues/134/comments","statuses_url":"https://api.github.com/repos/Azure/mmlspark/statuses/ad159a03d270a48a65bf0828e56cd4c3c23b33c6","head":{"label":"mhamilton723:streaming","ref":"streaming","sha":"ad159a03d270a48a65bf0828e56cd4c3c23b33c6","user":{"login":"mhamilton723","id":6456637,"avatar_url":"https://avatars2.githubusercontent.com/u/6456637?v=4","gravatar_id":"","url":"https://api.github.com/users/mhamilton723","html_url":"https://github.com/mhamilton723","followers_url":"https://api.github.com/users/mhamilton723/followers","following_url":"https://api.github.com/users/mhamilton723/following{/other_user}","gists_url":"https://api.github.com/users/mhamilton723/gists{/gist_id}","starred_url":"https://api.github.com/users/mhamilton723/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhamilton723/subscriptions","organizations_url":"https://api.github.com/users/mhamilton723/orgs","repos_url":"https://api.github.com/users/mhamilton723/repos","events_url":"https://api.github.com/users/mhamilton723/events{/privacy}","received_events_url":"https://api.github.com/users/mhamilton723/received_events","type":"User","site_admin":false},"repo":{"id":93436991,"name":"mmlspark","full_name":"mhamilton723/mmlspark","owner":{"login":"mhamilton723","id":6456637,"avatar_url":"https://avatars2.githubusercontent.com/u/6456637?v=4","gravatar_id":"","url":"https://api.github.com/users/mhamilton723","html_url":"https://github.com/mhamilton723","followers_url":"https://api.github.com/users/mhamilton723/followers","following_url":"https://api.github.com/users/mhamilton723/following{/other_user}","gists_url":"https://api.github.com/users/mhamilton723/gists{/gist_id}","starred_url":"https://api.github.com/users/mhamilton723/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhamilton723/subscriptions","organizations_url":"https://api.github.com/users/mhamilton723/orgs","repos_url":"https://api.github.com/users/mhamilton723/repos","events_url":"https://api.github.com/users/mhamilton723/events{/privacy}","received_events_url":"https://api.github.com/users/mhamilton723/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/mhamilton723/mmlspark","description":"Microsoft Machine Learning for Apache Spark","fork":true,"url":"https://api.github.com/repos/mhamilton723/mmlspark","forks_url":"https://api.github.com/repos/mhamilton723/mmlspark/forks","keys_url":"https://api.github.com/repos/mhamilton723/mmlspark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/mhamilton723/mmlspark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/mhamilton723/mmlspark/teams","hooks_url":"https://api.github.com/repos/mhamilton723/mmlspark/hooks","issue_events_url":"https://api.github.com/repos/mhamilton723/mmlspark/issues/events{/number}","events_url":"https://api.github.com/repos/mhamilton723/mmlspark/events","assignees_url":"https://api.github.com/repos/mhamilton723/mmlspark/assignees{/user}","branches_url":"https://api.github.com/repos/mhamilton723/mmlspark/branches{/branch}","tags_url":"https://api.github.com/repos/mhamilton723/mmlspark/tags","blobs_url":"https://api.github.com/repos/mhamilton723/mmlspark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/mhamilton723/mmlspark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/mhamilton723/mmlspark/git/refs{/sha}","trees_url":"https://api.github.com/repos/mhamilton723/mmlspark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/mhamilton723/mmlspark/statuses/{sha}","languages_url":"https://api.github.com/repos/mhamilton723/mmlspark/languages","stargazers_url":"https://api.github.com/repos/mhamilton723/mmlspark/stargazers","contributors_url":"https://api.github.com/repos/mhamilton723/mmlspark/contributors","subscribers_url":"https://api.github.com/repos/mhamilton723/mmlspark/subscribers","subscription_url":"https://api.github.com/repos/mhamilton723/mmlspark/subscription","commits_url":"https://api.github.com/repos/mhamilton723/mmlspark/commits{/sha}","git_commits_url":"https://api.github.com/repos/mhamilton723/mmlspark/git/commits{/sha}","comments_url":"https://api.github.com/repos/mhamilton723/mmlspark/comments{/number}","issue_comment_url":"https://api.github.com/repos/mhamilton723/mmlspark/issues/comments{/number}","contents_url":"https://api.github.com/repos/mhamilton723/mmlspark/contents/{+path}","compare_url":"https://api.github.com/repos/mhamilton723/mmlspark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/mhamilton723/mmlspark/merges","archive_url":"https://api.github.com/repos/mhamilton723/mmlspark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/mhamilton723/mmlspark/downloads","issues_url":"https://api.github.com/repos/mhamilton723/mmlspark/issues{/number}","pulls_url":"https://api.github.com/repos/mhamilton723/mmlspark/pulls{/number}","milestones_url":"https://api.github.com/repos/mhamilton723/mmlspark/milestones{/number}","notifications_url":"https://api.github.com/repos/mhamilton723/mmlspark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/mhamilton723/mmlspark/labels{/name}","releases_url":"https://api.github.com/repos/mhamilton723/mmlspark/releases{/id}","deployments_url":"https://api.github.com/repos/mhamilton723/mmlspark/deployments","created_at":"2017-06-05T19:00:34Z","updated_at":"2017-06-05T19:00:36Z","pushed_at":"2017-09-24T03:26:25Z","git_url":"git://github.com/mhamilton723/mmlspark.git","ssh_url":"git@github.com:mhamilton723/mmlspark.git","clone_url":"https://github.com/mhamilton723/mmlspark.git","svn_url":"https://github.com/mhamilton723/mmlspark","homepage":null,"size":2006,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"Azure:master","ref":"master","sha":"603df6d713d4bf392d619e0626eae3576444d7f3","user":{"login":"Azure","id":6844498,"avatar_url":"https://avatars0.githubusercontent.com/u/6844498?v=4","gravatar_id":"","url":"https://api.github.com/users/Azure","html_url":"https://github.com/Azure","followers_url":"https://api.github.com/users/Azure/followers","following_url":"https://api.github.com/users/Azure/following{/other_user}","gists_url":"https://api.github.com/users/Azure/gists{/gist_id}","starred_url":"https://api.github.com/users/Azure/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Azure/subscriptions","organizations_url":"https://api.github.com/users/Azure/orgs","repos_url":"https://api.github.com/users/Azure/repos","events_url":"https://api.github.com/users/Azure/events{/privacy}","received_events_url":"https://api.github.com/users/Azure/received_events","type":"Organization","site_admin":false},"repo":{"id":93381875,"name":"mmlspark","full_name":"Azure/mmlspark","owner":{"login":"Azure","id":6844498,"avatar_url":"https://avatars0.githubusercontent.com/u/6844498?v=4","gravatar_id":"","url":"https://api.github.com/users/Azure","html_url":"https://github.com/Azure","followers_url":"https://api.github.com/users/Azure/followers","following_url":"https://api.github.com/users/Azure/following{/other_user}","gists_url":"https://api.github.com/users/Azure/gists{/gist_id}","starred_url":"https://api.github.com/users/Azure/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Azure/subscriptions","organizations_url":"https://api.github.com/users/Azure/orgs","repos_url":"https://api.github.com/users/Azure/repos","events_url":"https://api.github.com/users/Azure/events{/privacy}","received_events_url":"https://api.github.com/users/Azure/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/Azure/mmlspark","description":"Microsoft Machine Learning for Apache Spark","fork":false,"url":"https://api.github.com/repos/Azure/mmlspark","forks_url":"https://api.github.com/repos/Azure/mmlspark/forks","keys_url":"https://api.github.com/repos/Azure/mmlspark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/Azure/mmlspark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/Azure/mmlspark/teams","hooks_url":"https://api.github.com/repos/Azure/mmlspark/hooks","issue_events_url":"https://api.github.com/repos/Azure/mmlspark/issues/events{/number}","events_url":"https://api.github.com/repos/Azure/mmlspark/events","assignees_url":"https://api.github.com/repos/Azure/mmlspark/assignees{/user}","branches_url":"https://api.github.com/repos/Azure/mmlspark/branches{/branch}","tags_url":"https://api.github.com/repos/Azure/mmlspark/tags","blobs_url":"https://api.github.com/repos/Azure/mmlspark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/Azure/mmlspark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/Azure/mmlspark/git/refs{/sha}","trees_url":"https://api.github.com/repos/Azure/mmlspark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/Azure/mmlspark/statuses/{sha}","languages_url":"https://api.github.com/repos/Azure/mmlspark/languages","stargazers_url":"https://api.github.com/repos/Azure/mmlspark/stargazers","contributors_url":"https://api.github.com/repos/Azure/mmlspark/contributors","subscribers_url":"https://api.github.com/repos/Azure/mmlspark/subscribers","subscription_url":"https://api.github.com/repos/Azure/mmlspark/subscription","commits_url":"https://api.github.com/repos/Azure/mmlspark/commits{/sha}","git_commits_url":"https://api.github.com/repos/Azure/mmlspark/git/commits{/sha}","comments_url":"https://api.github.com/repos/Azure/mmlspark/comments{/number}","issue_comment_url":"https://api.github.com/repos/Azure/mmlspark/issues/comments{/number}","contents_url":"https://api.github.com/repos/Azure/mmlspark/contents/{+path}","compare_url":"https://api.github.com/repos/Azure/mmlspark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/Azure/mmlspark/merges","archive_url":"https://api.github.com/repos/Azure/mmlspark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/Azure/mmlspark/downloads","issues_url":"https://api.github.com/repos/Azure/mmlspark/issues{/number}","pulls_url":"https://api.github.com/repos/Azure/mmlspark/pulls{/number}","milestones_url":"https://api.github.com/repos/Azure/mmlspark/milestones{/number}","notifications_url":"https://api.github.com/repos/Azure/mmlspark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/Azure/mmlspark/labels{/name}","releases_url":"https://api.github.com/repos/Azure/mmlspark/releases{/id}","deployments_url":"https://api.github.com/repos/Azure/mmlspark/deployments","created_at":"2017-06-05T08:23:44Z","updated_at":"2017-09-25T18:44:36Z","pushed_at":"2017-09-25T16:12:41Z","git_url":"git://github.com/Azure/mmlspark.git","ssh_url":"git@github.com:Azure/mmlspark.git","clone_url":"https://github.com/Azure/mmlspark.git","svn_url":"https://github.com/Azure/mmlspark","homepage":null,"size":631,"stargazers_count":210,"watchers_count":210,"language":"Scala","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":49,"mirror_url":null,"open_issues_count":22,"forks":49,"open_issues":22,"watchers":210,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/134"},"html":{"href":"https://github.com/Azure/mmlspark/pull/134"},"issue":{"href":"https://api.github.com/repos/Azure/mmlspark/issues/134"},"comments":{"href":"https://api.github.com/repos/Azure/mmlspark/issues/134/comments"},"review_comments":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/134/comments"},"review_comment":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/Azure/mmlspark/pulls/134/commits"},"statuses":{"href":"https://api.github.com/repos/Azure/mmlspark/statuses/ad159a03d270a48a65bf0828e56cd4c3c23b33c6"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"elibarzilay"}}