2141290691

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/JohnSnowLabs/spark-nlp/commit/<a class=hub.com/JohnSnowLabs/spark-nlp/commit/<a class="double-link" href="https://git"><a class=hub.com/JohnSnowLabs/spark-nlp/commit/<a class="double-link" href="https://git"><a class=hub.com/JohnSnowLabs/spark-nlp/commit/1b54a212be771f9031f09e6bc6100f2aa6627285">1b54a212b">&lt;a href=&quot;https://github.com/JohnSnowLabs/spark-nlp/commit/</a><a class="double-link" href="https://github.com/JohnSnowLabs/spark-nlp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/JohnSnowLabs/spark-nlp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/JohnSnowLabs/spark-nlp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/JohnSnowLabs/spark-nlp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/JohnSnowLabs/spark-nlp/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/JohnSnowLabs/spark-nlp/commit/1b54a212be771f9031f09e6bc6100f2aa6627285">1b54a212b</a><a href="https://github.com/JohnSnowLabs/spark-nlp/commit/1b54a212be771f9031f09e6bc6100f2aa6627285">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/JohnSnowLabs/spark-nlp/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/JohnSnowLabs/spark-nlp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/JohnSnowLabs/spark-nlp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/JohnSnowLabs/spark-nlp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/JohnSnowLabs/spark-nlp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/JohnSnowLabs/spark-nlp/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/JohnSnowLabs/spark-nlp/commit/1b54a212be771f9031f09e6bc6100f2aa6627285&quot;&gt;1b54a212b&lt;/a&gt;&lt;a href=&q... (continued)

Pull Request Pull Request #7234: Release/343 release candidate

Run Details

4001 of 4001 new or added lines in 186 files covered. (100.0%)

8081 of 10941 relevant lines covered (73.86%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.49

/src/main/scala/com/johnsnowlabs/nlp/pretrained/S3ResourceDownloader.scala

/*
 * Copyright 2017-2022 John Snow Labs
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.johnsnowlabs.nlp.pretrained

import com.johnsnowlabs.client.aws.AWSGateway
import com.johnsnowlabs.nlp.pretrained.ResourceDownloader.fileSystem
import com.johnsnowlabs.util.{ConfigHelper, ConfigLoader, FileHelper}
import org.apache.hadoop.fs.Path

import java.io.File
import java.nio.file.Files
import java.sql.Timestamp
import java.util.Calendar
import java.util.zip.ZipInputStream
import scala.collection.mutable

class S3ResourceDownloader(
    bucket: => String,
    s3Path: => String,
    cacheFolder: => String,
    credentialsType: => String,
    region: String = "us-east-1")
    extends ResourceDownloader {

  val repoFolder2Metadata: mutable.Map[String, RepositoryMetadata] =
    mutable.Map[String, RepositoryMetadata]()
  val cachePath = new Path(cacheFolder)

  if (!fileSystem.exists(cachePath)) {
    fileSystem.mkdirs(cachePath)
  }

  lazy val awsGateway = new AWSGateway(
    ConfigLoader.getConfigStringValue(ConfigHelper.accessKeyId),
    ConfigLoader.getConfigStringValue(ConfigHelper.secretAccessKey),
    ConfigLoader.getConfigStringValue(ConfigHelper.sessionToken),
    ConfigLoader.getConfigStringValue(ConfigHelper.awsProfileName),
    region,
    credentialsType)

  def downloadMetadataIfNeed(folder: String): List[ResourceMetadata] = {
    val lastState = repoFolder2Metadata.get(folder)

    val fiveMinsBefore = getTimestamp(-5)
    val needToRefresh = lastState.isEmpty || lastState.get.lastMetadataDownloaded
      .before(fiveMinsBefore)

    if (!needToRefresh) {
      lastState.get.metadata
    } else {
      awsGateway.getMetadata(s3Path, folder, bucket)
    }
  }

  def resolveLink(request: ResourceRequest): Option[ResourceMetadata] = {
    val metadata = downloadMetadataIfNeed(request.folder)
    ResourceMetadata.resolveResource(metadata, request)
  }

  /** Download resource to local file
    *
    * @param request
    *   Resource request
    * @return
    *   Downloaded file or None if resource is not found
    */
  override def download(request: ResourceRequest): Option[String] = {
    val link = resolveLink(request)
    link.flatMap { resource =>
      val s3FilePath = awsGateway.getS3File(s3Path, request.folder, resource.fileName)
      val destinationFile = new Path(cachePath.toString, resource.fileName)
      if (!awsGateway.doesS3ObjectExist(bucket, s3FilePath)) {
        None
      } else {
        downloadAndUnzipFile(destinationFile, resource, s3FilePath)
      }
    }
  }

  def downloadAndUnzipFile(
      destinationFile: Path,
      resource: ResourceMetadata,
      s3FilePath: String): Option[String] = {

    val splitPath = destinationFile.toString.substring(0, destinationFile.toString.length - 4)
    if (!(fileSystem.exists(destinationFile) || fileSystem.exists(new Path(splitPath)))) {
      // 1. Create tmp file
      val tmpFileName = Files.createTempFile(resource.fileName, "").toString
      val tmpFile = new File(tmpFileName)

      // 2. Download content to tmp file
      awsGateway.getS3Object(bucket, s3FilePath, tmpFile)
      // 3. validate checksum
      if (!resource.checksum.equals(""))
        require(
          FileHelper.generateChecksum(tmpFileName).equals(resource.checksum),
          "Checksum validation failed!")

      // 4. Move tmp file to destination
      fileSystem.moveFromLocalFile(new Path(tmpFile.toString), destinationFile)

    }

    // 5. Unzip if needs
    if (resource.isZipped) {
      // if not already unzipped
      if (!fileSystem.exists(new Path(splitPath))) {
        val zis = new ZipInputStream(fileSystem.open(destinationFile))
        val buf = Array.ofDim[Byte](1024)
        var entry = zis.getNextEntry
        require(
          destinationFile.toString.substring(destinationFile.toString.length - 4) == ".zip",
          "Not a zip file.")

        while (entry != null) {
          if (!entry.isDirectory) {
            val entryName = new Path(splitPath, entry.getName)
            val outputStream = fileSystem.create(entryName)
            var bytesRead = zis.read(buf, 0, 1024)
            while (bytesRead > -1) {
              outputStream.write(buf, 0, bytesRead)
              bytesRead = zis.read(buf, 0, 1024)
            }
            outputStream.close()
          }
          zis.closeEntry()
          entry = zis.getNextEntry
        }
        zis.close()
        // delete the zip file
        fileSystem.delete(destinationFile, true)
      }
      Some(splitPath)
    } else {
      Some(destinationFile.getName)
    }
  }

  def downloadAndUnzipFile(s3FilePath: String): Option[String] = {

    val s3File = s3FilePath.split("/").last
    val destinationFile = new Path(cachePath.toString + "/" + s3File)
    val splitPath = destinationFile.toString.substring(0, destinationFile.toString.length - 4)

    if (!(fileSystem.exists(destinationFile) || fileSystem.exists(new Path(splitPath)))) {
      // 1. Create tmp file
      val tmpFileName = Files.createTempFile(s3File, "").toString
      val tmpFile = new File(tmpFileName)

      val newStrfilePath: String = s3FilePath.toString
      val mybucket: String = bucket.toString
      // 2. Download content to tmp file
      awsGateway.getS3Object(mybucket, newStrfilePath, tmpFile)

      // 4. Move tmp file to destination
      fileSystem.moveFromLocalFile(new Path(tmpFile.toString), destinationFile)
    }

    if (!fileSystem.exists(new Path(splitPath))) {
      val zis = new ZipInputStream(fileSystem.open(destinationFile))
      val buf = Array.ofDim[Byte](1024)
      var entry = zis.getNextEntry
      require(
        destinationFile.toString.substring(destinationFile.toString.length - 4) == ".zip",
        "Not a zip file.")

      while (entry != null) {
        if (!entry.isDirectory) {
          val entryName = new Path(splitPath, entry.getName)
          val outputStream = fileSystem.create(entryName)
          var bytesRead = zis.read(buf, 0, 1024)
          while (bytesRead > -1) {
            outputStream.write(buf, 0, bytesRead)
            bytesRead = zis.read(buf, 0, 1024)
          }
          outputStream.close()
        }
        zis.closeEntry()
        entry = zis.getNextEntry
      }
      zis.close()
      // delete the zip file
      fileSystem.delete(destinationFile, true)
    }
    Some(splitPath)

  }

  override def getDownloadSize(request: ResourceRequest): Option[Long] = {
    val link = resolveLink(request)
    link.flatMap { resource =>
      awsGateway.getS3DownloadSize(s3Path, request.folder, resource.fileName, bucket)
    }
  }

  override def clearCache(request: ResourceRequest): Unit = {
    val metadata = downloadMetadataIfNeed(request.folder)

    val resources = ResourceMetadata.resolveResource(metadata, request)
    for (resource <- resources) {
      val fileName = new Path(cachePath.toString, resource.fileName)
      if (fileSystem.exists(fileName))
        fileSystem.delete(fileName, true)

      if (resource.isZipped) {
        require(fileName.toString.substring(fileName.toString.length - 4) == ".zip")
        val unzipped = fileName.toString.substring(0, fileName.toString.length - 4)
        val unzippedFile = new Path(unzipped)
        if (fileSystem.exists(unzippedFile))
          fileSystem.delete(unzippedFile, true)
      }
    }
  }

  private def getTimestamp(addMinutes: Int = 0): Timestamp = {
    val cal = Calendar.getInstance()
    cal.add(Calendar.MINUTE, addMinutes)
    val timestamp = new Timestamp(cal.getTime.getTime)
    cal.clear()
    timestamp
  }

}

1	/*
2	* Copyright 2017-2022 John Snow Labs
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package com.johnsnowlabs.nlp.pretrained
18
19	import com.johnsnowlabs.client.aws.AWSGateway
20	import com.johnsnowlabs.nlp.pretrained.ResourceDownloader.fileSystem
21	import com.johnsnowlabs.util.{ConfigHelper, ConfigLoader, FileHelper}
22	import org.apache.hadoop.fs.Path
23
24	import java.io.File
25	import java.nio.file.Files
26	import java.sql.Timestamp
27	import java.util.Calendar
28	import java.util.zip.ZipInputStream
29	import scala.collection.mutable
30
31	class S3ResourceDownloader(
32	bucket: => String,
33	s3Path: => String,
34	cacheFolder: => String,
35	credentialsType: => String,
36	region: String = "us-east-1")
37	extends ResourceDownloader {
38
39	val repoFolder2Metadata: mutable.Map[String, RepositoryMetadata] =
40	mutable.Map[String, RepositoryMetadata]()	1✔
41	val cachePath = new Path(cacheFolder)	1✔
42
43	if (!fileSystem.exists(cachePath)) {	1✔
44	fileSystem.mkdirs(cachePath)	1✔
45	}
46
47	lazy val awsGateway = new AWSGateway(
48	ConfigLoader.getConfigStringValue(ConfigHelper.accessKeyId),
49	ConfigLoader.getConfigStringValue(ConfigHelper.secretAccessKey),
50	ConfigLoader.getConfigStringValue(ConfigHelper.sessionToken),
51	ConfigLoader.getConfigStringValue(ConfigHelper.awsProfileName),
52	region,
53	credentialsType)
54
55	def downloadMetadataIfNeed(folder: String): List[ResourceMetadata] = {
56	val lastState = repoFolder2Metadata.get(folder)	1✔
57
58	val fiveMinsBefore = getTimestamp(-5)	1✔
59	val needToRefresh = lastState.isEmpty \|\| lastState.get.lastMetadataDownloaded	1✔
60	.before(fiveMinsBefore)	×
61
62	if (!needToRefresh) {	1✔
63	lastState.get.metadata	×
64	} else {
65	awsGateway.getMetadata(s3Path, folder, bucket)	1✔
66	}
67	}
68
69	def resolveLink(request: ResourceRequest): Option[ResourceMetadata] = {
70	val metadata = downloadMetadataIfNeed(request.folder)	1✔
71	ResourceMetadata.resolveResource(metadata, request)	1✔
72	}
73
74	/** Download resource to local file
75	*
76	* @param request
77	* Resource request
78	* @return
79	* Downloaded file or None if resource is not found
80	*/
81	override def download(request: ResourceRequest): Option[String] = {
82	val link = resolveLink(request)	1✔
83	link.flatMap { resource =>	1✔
84	val s3FilePath = awsGateway.getS3File(s3Path, request.folder, resource.fileName)	1✔
85	val destinationFile = new Path(cachePath.toString, resource.fileName)	1✔
86	if (!awsGateway.doesS3ObjectExist(bucket, s3FilePath)) {	1✔
87	None	×
88	} else {
89	downloadAndUnzipFile(destinationFile, resource, s3FilePath)	1✔
90	}
91	}
92	}
93
94	def downloadAndUnzipFile(
95	destinationFile: Path,
96	resource: ResourceMetadata,
97	s3FilePath: String): Option[String] = {
98
99	val splitPath = destinationFile.toString.substring(0, destinationFile.toString.length - 4)	1✔
100	if (!(fileSystem.exists(destinationFile) \|\| fileSystem.exists(new Path(splitPath)))) {	1✔
101	// 1. Create tmp file
102	val tmpFileName = Files.createTempFile(resource.fileName, "").toString	1✔
103	val tmpFile = new File(tmpFileName)	1✔
104
105	// 2. Download content to tmp file
106	awsGateway.getS3Object(bucket, s3FilePath, tmpFile)	1✔
107	// 3. validate checksum
108	if (!resource.checksum.equals(""))	1✔
109	require(	×
110	FileHelper.generateChecksum(tmpFileName).equals(resource.checksum),	×
111	"Checksum validation failed!")	×
112
113	// 4. Move tmp file to destination
114	fileSystem.moveFromLocalFile(new Path(tmpFile.toString), destinationFile)	1✔
115
116	}
117
118	// 5. Unzip if needs
119	if (resource.isZipped) {	1✔
120	// if not already unzipped
121	if (!fileSystem.exists(new Path(splitPath))) {	1✔
122	val zis = new ZipInputStream(fileSystem.open(destinationFile))	1✔
123	val buf = Array.ofDim[Byte](1024)	1✔
124	var entry = zis.getNextEntry	1✔
125	require(	1✔
126	destinationFile.toString.substring(destinationFile.toString.length - 4) == ".zip",	1✔
127	"Not a zip file.")	×
128
129	while (entry != null) {	1✔
130	if (!entry.isDirectory) {	1✔
131	val entryName = new Path(splitPath, entry.getName)	1✔
132	val outputStream = fileSystem.create(entryName)	1✔
133	var bytesRead = zis.read(buf, 0, 1024)	1✔
134	while (bytesRead > -1) {	1✔
135	outputStream.write(buf, 0, bytesRead)	1✔
136	bytesRead = zis.read(buf, 0, 1024)	1✔
137	}
138	outputStream.close()	1✔
139	}
140	zis.closeEntry()	1✔
141	entry = zis.getNextEntry	1✔
142	}
143	zis.close()	1✔
144	// delete the zip file
145	fileSystem.delete(destinationFile, true)	1✔
146	}
147	Some(splitPath)	1✔
148	} else {
149	Some(destinationFile.getName)	×
150	}
151	}
152
153	def downloadAndUnzipFile(s3FilePath: String): Option[String] = {
154
155	val s3File = s3FilePath.split("/").last	×
156	val destinationFile = new Path(cachePath.toString + "/" + s3File)	×
157	val splitPath = destinationFile.toString.substring(0, destinationFile.toString.length - 4)	×
158
159	if (!(fileSystem.exists(destinationFile) \|\| fileSystem.exists(new Path(splitPath)))) {	×
160	// 1. Create tmp file
161	val tmpFileName = Files.createTempFile(s3File, "").toString	×
162	val tmpFile = new File(tmpFileName)	×
163
164	val newStrfilePath: String = s3FilePath.toString	×
165	val mybucket: String = bucket.toString	×
166	// 2. Download content to tmp file
167	awsGateway.getS3Object(mybucket, newStrfilePath, tmpFile)	×
168
169	// 4. Move tmp file to destination
170	fileSystem.moveFromLocalFile(new Path(tmpFile.toString), destinationFile)	×
171	}
172
173	if (!fileSystem.exists(new Path(splitPath))) {	×
174	val zis = new ZipInputStream(fileSystem.open(destinationFile))	×
175	val buf = Array.ofDim[Byte](1024)	×
176	var entry = zis.getNextEntry	×
177	require(	×
178	destinationFile.toString.substring(destinationFile.toString.length - 4) == ".zip",	×
179	"Not a zip file.")	×
180
181	while (entry != null) {	×
182	if (!entry.isDirectory) {	×
183	val entryName = new Path(splitPath, entry.getName)	×
184	val outputStream = fileSystem.create(entryName)	×
185	var bytesRead = zis.read(buf, 0, 1024)	×
186	while (bytesRead > -1) {	×
187	outputStream.write(buf, 0, bytesRead)	×
188	bytesRead = zis.read(buf, 0, 1024)	×
189	}
190	outputStream.close()	×
191	}
192	zis.closeEntry()	×
193	entry = zis.getNextEntry	×
194	}
195	zis.close()	×
196	// delete the zip file
197	fileSystem.delete(destinationFile, true)	×
198	}
199	Some(splitPath)	×
200
201	}
202
203	override def getDownloadSize(request: ResourceRequest): Option[Long] = {
204	val link = resolveLink(request)	1✔
205	link.flatMap { resource =>	1✔
206	awsGateway.getS3DownloadSize(s3Path, request.folder, resource.fileName, bucket)	1✔
207	}
208	}
209
210	override def clearCache(request: ResourceRequest): Unit = {
211	val metadata = downloadMetadataIfNeed(request.folder)	×
212
213	val resources = ResourceMetadata.resolveResource(metadata, request)	×
214	for (resource <- resources) {	×
215	val fileName = new Path(cachePath.toString, resource.fileName)	×
216	if (fileSystem.exists(fileName))	×
217	fileSystem.delete(fileName, true)	×
218
219	if (resource.isZipped) {	×
220	require(fileName.toString.substring(fileName.toString.length - 4) == ".zip")	×
221	val unzipped = fileName.toString.substring(0, fileName.toString.length - 4)	×
222	val unzippedFile = new Path(unzipped)	×
223	if (fileSystem.exists(unzippedFile))	×
224	fileSystem.delete(unzippedFile, true)	×
225	}
226	}
227	}
228
229	private def getTimestamp(addMinutes: Int = 0): Timestamp = {
230	val cal = Calendar.getInstance()	1✔
231	cal.add(Calendar.MINUTE, addMinutes)	1✔
232	val timestamp = new Timestamp(cal.getTime.getTime)	1✔
233	cal.clear()	1✔
234	timestamp
235	}
236
237	}

JohnSnowLabs / spark-nlp / 2141290691

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous