From 890a845184bbd0e3d60068c1fb65874b89baaf63 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 17:37:46 +0800 Subject: [PATCH 01/11] init pr --- .../apache/spark/ml/image/ImageSchema.scala | 266 ------------------ .../ml/source/image/ImageFileFormat.scala | 50 +++- .../spark/ml/source/image/ImageUtils.scala | 118 ++++++++ .../spark/ml/image/ImageSchemaSuite.scala | 171 ----------- .../source/image/ImageFileFormatSuite.scala | 4 +- python/pyspark/ml/image.py | 50 +--- python/pyspark/ml/tests/test_image.py | 34 ++- 7 files changed, 188 insertions(+), 505 deletions(-) delete mode 100644 mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala create mode 100644 mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala delete mode 100644 mllib/src/test/scala/org/apache/spark/ml/image/ImageSchemaSuite.scala diff --git a/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala b/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala deleted file mode 100644 index a7ddf2f1f130f..0000000000000 --- a/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala +++ /dev/null @@ -1,266 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.spark.ml.image - -import java.awt.Color -import java.awt.color.ColorSpace -import java.io.ByteArrayInputStream -import javax.imageio.ImageIO - -import scala.collection.JavaConverters._ - -import org.apache.spark.annotation.{Experimental, Since} -import org.apache.spark.input.PortableDataStream -import org.apache.spark.sql.{DataFrame, Row, SparkSession} -import org.apache.spark.sql.types._ - -/** - * :: Experimental :: - * Defines the image schema and methods to read and manipulate images. - */ -@Experimental -@Since("2.3.0") -object ImageSchema { - - val undefinedImageType = "Undefined" - - /** - * (Scala-specific) OpenCV type mapping supported - */ - val ocvTypes: Map[String, Int] = Map( - undefinedImageType -> -1, - "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 - ) - - /** - * (Java-specific) OpenCV type mapping supported - */ - val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava - - /** - * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) - */ - val columnSchema = StructType( - StructField("origin", StringType, true) :: - StructField("height", IntegerType, false) :: - StructField("width", IntegerType, false) :: - StructField("nChannels", IntegerType, false) :: - // OpenCV-compatible type: CV_8UC3 in most cases - StructField("mode", IntegerType, false) :: - // Bytes in OpenCV-compatible order: row-wise BGR in most cases - StructField("data", BinaryType, false) :: Nil) - - val imageFields: Array[String] = columnSchema.fieldNames - - /** - * DataFrame with a single column of images named "image" (nullable) - */ - val imageSchema = StructType(StructField("image", columnSchema, true) :: Nil) - - /** - * Gets the origin of the image - * - * @return The origin of the image - */ - def getOrigin(row: Row): String = row.getString(0) - - /** - * Gets the height of the image - * - * @return The height of the image - */ - def getHeight(row: Row): Int = row.getInt(1) - - /** - * Gets the width of the image - * - * @return The width of the image - */ - def getWidth(row: Row): Int = row.getInt(2) - - /** - * Gets the number of channels in the image - * - * @return The number of channels in the image - */ - def getNChannels(row: Row): Int = row.getInt(3) - - /** - * Gets the OpenCV representation as an int - * - * @return The OpenCV representation as an int - */ - def getMode(row: Row): Int = row.getInt(4) - - /** - * Gets the image data - * - * @return The image data - */ - def getData(row: Row): Array[Byte] = row.getAs[Array[Byte]](5) - - /** - * Default values for the invalid image - * - * @param origin Origin of the invalid image - * @return Row with the default values - */ - private[spark] def invalidImageRow(origin: String): Row = - Row(Row(origin, -1, -1, -1, ocvTypes(undefinedImageType), Array.ofDim[Byte](0))) - - /** - * Convert the compressed image (jpeg, png, etc.) into OpenCV - * representation and store it in DataFrame Row - * - * @param origin Arbitrary string that identifies the image - * @param bytes Image bytes (for example, jpeg) - * @return DataFrame Row or None (if the decompression fails) - */ - private[spark] def decode(origin: String, bytes: Array[Byte]): Option[Row] = { - - val img = try { - ImageIO.read(new ByteArrayInputStream(bytes)) - } catch { - // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. - // But do not catch the declared `IOException` (regarded as FileSystem failure) - case _: RuntimeException => null - } - - if (img == null) { - None - } else { - val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY - val hasAlpha = img.getColorModel.hasAlpha - - val height = img.getHeight - val width = img.getWidth - val (nChannels, mode) = if (isGray) { - (1, ocvTypes("CV_8UC1")) - } else if (hasAlpha) { - (4, ocvTypes("CV_8UC4")) - } else { - (3, ocvTypes("CV_8UC3")) - } - - val imageSize = height * width * nChannels - assert(imageSize < 1e9, "image is too large") - val decoded = Array.ofDim[Byte](imageSize) - - // Grayscale images in Java require special handling to get the correct intensity - if (isGray) { - var offset = 0 - val raster = img.getRaster - for (h <- 0 until height) { - for (w <- 0 until width) { - decoded(offset) = raster.getSample(w, h, 0).toByte - offset += 1 - } - } - } else { - var offset = 0 - for (h <- 0 until height) { - for (w <- 0 until width) { - val color = new Color(img.getRGB(w, h), hasAlpha) - decoded(offset) = color.getBlue.toByte - decoded(offset + 1) = color.getGreen.toByte - decoded(offset + 2) = color.getRed.toByte - if (hasAlpha) { - decoded(offset + 3) = color.getAlpha.toByte - } - offset += nChannels - } - } - } - - // the internal "Row" is needed, because the image is a single DataFrame column - Some(Row(Row(origin, height, width, nChannels, mode, decoded))) - } - } - - /** - * Read the directory of images from the local or remote source - * - * @note If multiple jobs are run in parallel with different sampleRatio or recursive flag, - * there may be a race condition where one job overwrites the hadoop configs of another. - * @note If sample ratio is less than 1, sampling uses a PathFilter that is efficient but - * potentially non-deterministic. - * - * @param path Path to the image directory - * @return DataFrame with a single column "image" of images; - * see ImageSchema for the details - */ - @deprecated("use `spark.read.format(\"image\").load(path)` and this `readImages` will be " + - "removed in 3.0.0.", "2.4.0") - def readImages(path: String): DataFrame = readImages(path, null, false, -1, false, 1.0, 0) - - /** - * Read the directory of images from the local or remote source - * - * @note If multiple jobs are run in parallel with different sampleRatio or recursive flag, - * there may be a race condition where one job overwrites the hadoop configs of another. - * @note If sample ratio is less than 1, sampling uses a PathFilter that is efficient but - * potentially non-deterministic. - * - * @param path Path to the image directory - * @param sparkSession Spark Session, if omitted gets or creates the session - * @param recursive Recursive path search flag - * @param numPartitions Number of the DataFrame partitions, - * if omitted uses defaultParallelism instead - * @param dropImageFailures Drop the files that are not valid images from the result - * @param sampleRatio Fraction of the files loaded - * @return DataFrame with a single column "image" of images; - * see ImageSchema for the details - */ - @deprecated("use `spark.read.format(\"image\").load(path)` and this `readImages` will be " + - "removed in 3.0.0.", "2.4.0") - def readImages( - path: String, - sparkSession: SparkSession, - recursive: Boolean, - numPartitions: Int, - dropImageFailures: Boolean, - sampleRatio: Double, - seed: Long): DataFrame = { - require(sampleRatio <= 1.0 && sampleRatio >= 0, "sampleRatio should be between 0 and 1") - - val session = if (sparkSession != null) sparkSession else SparkSession.builder().getOrCreate - val partitions = - if (numPartitions > 0) { - numPartitions - } else { - session.sparkContext.defaultParallelism - } - - RecursiveFlag.withRecursiveFlag(recursive, session) { - SamplePathFilter.withPathFilter(sampleRatio, session, seed) { - val binResult = session.sparkContext.binaryFiles(path, partitions) - val streams = if (numPartitions == -1) binResult else binResult.repartition(partitions) - val convert = (origin: String, bytes: PortableDataStream) => - decode(origin, bytes.toArray()) - val images = if (dropImageFailures) { - streams.flatMap { case (origin, bytes) => convert(origin, bytes) } - } else { - streams.map { case (origin, bytes) => - convert(origin, bytes).getOrElse(invalidImageRow(origin)) - } - } - session.createDataFrame(images, imageSchema) - } - } - } -} diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index c3321447e3c96..1a10c150896eb 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -22,23 +22,23 @@ import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.mapreduce.Job -import org.apache.spark.ml.image.ImageSchema -import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.encoders.RowEncoder -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, UnsafeRow} -import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap -import org.apache.spark.sql.execution.datasources.{DataSource, FileFormat, OutputWriterFactory, PartitionedFile} +import org.apache.spark.sql.catalyst.expressions.UnsafeRow +import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, + PartitionedFile} import org.apache.spark.sql.sources.{DataSourceRegister, Filter} -import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.types._ import org.apache.spark.util.SerializableConfiguration private[image] class ImageFileFormat extends FileFormat with DataSourceRegister { + import ImageFileFormat._ override def inferSchema( sparkSession: SparkSession, options: Map[String, String], - files: Seq[FileStatus]): Option[StructType] = Some(ImageSchema.imageSchema) + files: Seq[FileStatus]): Option[StructType] = Some(schema) override def prepareWrite( sparkSession: SparkSession, @@ -81,11 +81,11 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } finally { Closeables.close(stream, true) } - val resultOpt = ImageSchema.decode(origin, bytes) + val resultOpt = ImageUtils.decode(origin, bytes) val filteredResult = if (imageSourceOptions.dropInvalid) { resultOpt.toIterator } else { - Iterator(resultOpt.getOrElse(ImageSchema.invalidImageRow(origin))) + Iterator(resultOpt.getOrElse(invalidImageRow(origin))) } if (requiredSchema.isEmpty) { @@ -98,3 +98,35 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } } } + +object ImageFileFormat { + /** + * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) + */ + private[image] val columnSchema = StructType( + StructField("origin", StringType, true) :: + StructField("height", IntegerType, false) :: + StructField("width", IntegerType, false) :: + StructField("nChannels", IntegerType, false) :: + // OpenCV-compatible type: CV_8UC3 in most cases + StructField("mode", IntegerType, false) :: + // Bytes in OpenCV-compatible order: row-wise BGR in most cases + StructField("data", BinaryType, false) :: Nil) + + private[image] val imageFields: Array[String] = columnSchema.fieldNames + + /** + * DataFrame with a single column of images named "image" (nullable) + */ + val schema = StructType(StructField("image", columnSchema, true) :: Nil) + + /** + * Default values for the invalid image + * + * @param origin Origin of the invalid image + * @return Row with the default values + */ + private[image] def invalidImageRow(origin: String): Row = + Row(Row(origin, -1, -1, -1, ImageUtils.ocvTypes(ImageUtils.undefinedImageType), + Array.ofDim[Byte](0))) +} diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala new file mode 100644 index 0000000000000..95ae8f68a43da --- /dev/null +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.spark.ml.source.image + +import java.awt.Color +import java.awt.color.ColorSpace +import java.io.ByteArrayInputStream +import javax.imageio.ImageIO + +import scala.collection.JavaConverters._ + +import org.apache.spark.sql.Row + +/** + * Defines the image util methods to read and manipulate images. + */ +private[spark] object ImageUtils { + + val undefinedImageType = "Undefined" + + /** + * (Scala-specific) OpenCV type mapping supported + */ + val ocvTypes: Map[String, Int] = Map( + undefinedImageType -> -1, + "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 + ) + + /** + * (Java-specific) OpenCV type mapping supported + */ + val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava + + /** + * Convert the compressed image (jpeg, png, etc.) into OpenCV + * representation and store it in DataFrame Row + * + * @param origin Arbitrary string that identifies the image + * @param bytes Image bytes (for example, jpeg) + * @return DataFrame Row or None (if the decompression fails) + */ + def decode(origin: String, bytes: Array[Byte]): Option[Row] = { + + val img = try { + ImageIO.read(new ByteArrayInputStream(bytes)) + } catch { + // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. + // But do not catch the declared `IOException` (regarded as FileSystem failure) + case _: RuntimeException => null + } + + if (img == null) { + None + } else { + val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY + val hasAlpha = img.getColorModel.hasAlpha + + val height = img.getHeight + val width = img.getWidth + val (nChannels, mode) = if (isGray) { + (1, ocvTypes("CV_8UC1")) + } else if (hasAlpha) { + (4, ocvTypes("CV_8UC4")) + } else { + (3, ocvTypes("CV_8UC3")) + } + + val imageSize = height * width * nChannels + assert(imageSize < 1e9, "image is too large") + val decoded = Array.ofDim[Byte](imageSize) + + // Grayscale images in Java require special handling to get the correct intensity + if (isGray) { + var offset = 0 + val raster = img.getRaster + for (h <- 0 until height) { + for (w <- 0 until width) { + decoded(offset) = raster.getSample(w, h, 0).toByte + offset += 1 + } + } + } else { + var offset = 0 + for (h <- 0 until height) { + for (w <- 0 until width) { + val color = new Color(img.getRGB(w, h), hasAlpha) + decoded(offset) = color.getBlue.toByte + decoded(offset + 1) = color.getGreen.toByte + decoded(offset + 2) = color.getRed.toByte + if (hasAlpha) { + decoded(offset + 3) = color.getAlpha.toByte + } + offset += nChannels + } + } + } + + // the internal "Row" is needed, because the image is a single DataFrame column + Some(Row(Row(origin, height, width, nChannels, mode, decoded))) + } + } + +} diff --git a/mllib/src/test/scala/org/apache/spark/ml/image/ImageSchemaSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/image/ImageSchemaSuite.scala deleted file mode 100644 index e16ec906c90b1..0000000000000 --- a/mllib/src/test/scala/org/apache/spark/ml/image/ImageSchemaSuite.scala +++ /dev/null @@ -1,171 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.spark.ml.image - -import java.nio.file.Paths -import java.util.Arrays - -import org.apache.spark.SparkFunSuite -import org.apache.spark.ml.image.ImageSchema._ -import org.apache.spark.mllib.util.MLlibTestSparkContext -import org.apache.spark.sql.Row -import org.apache.spark.sql.types._ - -class ImageSchemaSuite extends SparkFunSuite with MLlibTestSparkContext { - // Single column of images named "image" - private lazy val imagePath = "../data/mllib/images/origin" - - test("Smoke test: create basic ImageSchema dataframe") { - val origin = "path" - val width = 1 - val height = 1 - val nChannels = 3 - val data = Array[Byte](0, 0, 0) - val mode = ocvTypes("CV_8UC3") - - // Internal Row corresponds to image StructType - val rows = Seq(Row(Row(origin, height, width, nChannels, mode, data)), - Row(Row(null, height, width, nChannels, mode, data))) - val rdd = sc.makeRDD(rows) - val df = spark.createDataFrame(rdd, ImageSchema.imageSchema) - - assert(df.count === 2, "incorrect image count") - assert(df.schema("image").dataType == columnSchema, "data do not fit ImageSchema") - } - - test("readImages count test") { - var df = readImages(imagePath) - assert(df.count === 1) - - df = readImages(imagePath, null, true, -1, false, 1.0, 0) - assert(df.count === 10) - - df = readImages(imagePath, null, true, -1, true, 1.0, 0) - val countTotal = df.count - assert(countTotal === 8) - - df = readImages(imagePath, null, true, -1, true, 0.5, 0) - // Random number about half of the size of the original dataset - val count50 = df.count - assert(count50 > 0 && count50 < countTotal) - } - - test("readImages test: recursive = false") { - val df = readImages(imagePath, null, false, 3, true, 1.0, 0) - assert(df.count() === 0) - } - - test("readImages test: read jpg image") { - val df = readImages(imagePath + "/kittens/DP153539.jpg", null, false, 3, true, 1.0, 0) - assert(df.count() === 1) - } - - test("readImages test: read png image") { - val df = readImages(imagePath + "/multi-channel/BGRA.png", null, false, 3, true, 1.0, 0) - assert(df.count() === 1) - } - - test("readImages test: read non image") { - val df = readImages(imagePath + "/kittens/not-image.txt", null, false, 3, true, 1.0, 0) - assert(df.schema("image").dataType == columnSchema, "data do not fit ImageSchema") - assert(df.count() === 0) - } - - test("readImages test: read non image and dropImageFailures is false") { - val df = readImages(imagePath + "/kittens/not-image.txt", null, false, 3, false, 1.0, 0) - assert(df.count() === 1) - } - - test("readImages test: sampleRatio > 1") { - val e = intercept[IllegalArgumentException] { - readImages(imagePath, null, true, 3, true, 1.1, 0) - } - assert(e.getMessage.contains("sampleRatio")) - } - - test("readImages test: sampleRatio < 0") { - val e = intercept[IllegalArgumentException] { - readImages(imagePath, null, true, 3, true, -0.1, 0) - } - assert(e.getMessage.contains("sampleRatio")) - } - - test("readImages test: sampleRatio = 0") { - val df = readImages(imagePath, null, true, 3, true, 0.0, 0) - assert(df.count() === 0) - } - - test("readImages test: with sparkSession") { - val df = readImages(imagePath, sparkSession = spark, true, 3, true, 1.0, 0) - assert(df.count() === 8) - } - - test("readImages partition test") { - val df = readImages(imagePath, null, true, 3, true, 1.0, 0) - assert(df.rdd.getNumPartitions === 3) - } - - test("readImages partition test: < 0") { - val df = readImages(imagePath, null, true, -3, true, 1.0, 0) - assert(df.rdd.getNumPartitions === spark.sparkContext.defaultParallelism) - } - - test("readImages partition test: = 0") { - val df = readImages(imagePath, null, true, 0, true, 1.0, 0) - assert(df.rdd.getNumPartitions === spark.sparkContext.defaultParallelism) - } - - // Images with the different number of channels - test("readImages pixel values test") { - - val images = readImages(imagePath + "/multi-channel/").collect - - images.foreach { rrow => - val row = rrow.getAs[Row](0) - val filename = Paths.get(getOrigin(row)).getFileName().toString() - if (firstBytes20.contains(filename)) { - val mode = getMode(row) - val bytes20 = getData(row).slice(0, 20) - - val (expectedMode, expectedBytes) = firstBytes20(filename) - assert(ocvTypes(expectedMode) === mode, "mode of the image is not read correctly") - assert(Arrays.equals(expectedBytes, bytes20), "incorrect numeric value for flattened image") - } - } - } - - // number of channels and first 20 bytes of OpenCV representation - // - default representation for 3-channel RGB images is BGR row-wise: - // (B00, G00, R00, B10, G10, R10, ...) - // - default representation for 4-channel RGB images is BGRA row-wise: - // (B00, G00, R00, A00, B10, G10, R10, A10, ...) - private val firstBytes20 = Map( - "grayscale.jpg" -> - (("CV_8UC1", Array[Byte](-2, -33, -61, -60, -59, -59, -64, -59, -66, -67, -73, -73, -62, - -57, -60, -63, -53, -49, -55, -69))), - "chr30.4.184.jpg" -> (("CV_8UC3", - Array[Byte](-9, -3, -1, -43, -32, -28, -75, -60, -57, -78, -59, -56, -74, -59, -57, - -71, -58, -56, -73, -64))), - "BGRA.png" -> (("CV_8UC4", - Array[Byte](-128, -128, -8, -1, -128, -128, -8, -1, -128, - -128, -8, -1, 127, 127, -9, -1, 127, 127, -9, -1))), - "BGRA_alpha_60.png" -> (("CV_8UC4", - Array[Byte](-128, -128, -8, 60, -128, -128, -8, 60, -128, - -128, -8, 60, 127, 127, -9, 60, 127, 127, -9, 60))) - ) -} diff --git a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala index 38bb246d02184..3f94918d14832 100644 --- a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala @@ -21,7 +21,7 @@ import java.net.URI import java.nio.file.Paths import org.apache.spark.SparkFunSuite -import org.apache.spark.ml.image.ImageSchema._ +import org.apache.spark.ml.source.image.ImageUtils._ import org.apache.spark.mllib.util.MLlibTestSparkContext import org.apache.spark.sql.Row import org.apache.spark.sql.functions.{col, substring_index} @@ -67,7 +67,7 @@ class ImageFileFormatSuite extends SparkFunSuite with MLlibTestSparkContext { assert(new URI(resultOrigin) === Paths.get(filePath).toAbsolutePath().normalize().toUri()) // Compare other columns in the row to be the same with the `invalidImageRow` - assert(result === invalidImageRow(resultOrigin)) + assert(result === ImageFileFormat.invalidImageRow(resultOrigin)) } test("image datasource partition test") { diff --git a/python/pyspark/ml/image.py b/python/pyspark/ml/image.py index a1aacea88e42e..13e278e9a06e0 100644 --- a/python/pyspark/ml/image.py +++ b/python/pyspark/ml/image.py @@ -34,10 +34,10 @@ from pyspark.sql.types import Row, _create_row, _parse_datatype_json_string from pyspark.sql import DataFrame, SparkSession -__all__ = ["ImageSchema"] +__all__ = ["ImageUtils"] -class _ImageSchema(object): +class _ImageUtils(object): """ Internal class for `pyspark.ml.image.ImageSchema` attribute. Meant to be private and not to be instantized. Use `pyspark.ml.image.ImageSchema` attribute to access the @@ -178,7 +178,7 @@ def toImage(self, array, origin=""): raise ValueError("Invalid array shape") height, width, nChannels = array.shape - ocvTypes = ImageSchema.ocvTypes + ocvTypes = ImageUtils.ocvTypes if nChannels == 1: mode = ocvTypes["CV_8UC1"] elif nChannels == 3: @@ -203,52 +203,16 @@ def toImage(self, array, origin=""): return _create_row(self.imageFields, [origin, height, width, nChannels, mode, data]) - def readImages(self, path, recursive=False, numPartitions=-1, - dropImageFailures=False, sampleRatio=1.0, seed=0): - """ - Reads the directory of images from the local or remote source. - - .. note:: If multiple jobs are run in parallel with different sampleRatio or recursive flag, - there may be a race condition where one job overwrites the hadoop configs of another. - - .. note:: If sample ratio is less than 1, sampling uses a PathFilter that is efficient but - potentially non-deterministic. - - .. note:: Deprecated in 2.4.0. Use `spark.read.format("image").load(path)` instead and - this `readImages` will be removed in 3.0.0. - - :param str path: Path to the image directory. - :param bool recursive: Recursive search flag. - :param int numPartitions: Number of DataFrame partitions. - :param bool dropImageFailures: Drop the files that are not valid images. - :param float sampleRatio: Fraction of the images loaded. - :param int seed: Random number seed. - :return: a :class:`DataFrame` with a single column of "images", - see ImageSchema for details. - - >>> df = ImageSchema.readImages('data/mllib/images/origin/kittens', recursive=True) - >>> df.count() - 5 - - .. versionadded:: 2.3.0 - """ - warnings.warn("`ImageSchema.readImage` is deprecated. " + - "Use `spark.read.format(\"image\").load(path)` instead.", DeprecationWarning) - spark = SparkSession.builder.getOrCreate() - image_schema = spark._jvm.org.apache.spark.ml.image.ImageSchema - jsession = spark._jsparkSession - jresult = image_schema.readImages(path, jsession, recursive, numPartitions, - dropImageFailures, float(sampleRatio), seed) - return DataFrame(jresult, spark._wrapped) - -ImageSchema = _ImageSchema() +ImageUtils = _ImageUtils() # Monkey patch to disallow instantiation of this class. def _disallow_instance(_): raise RuntimeError("Creating instance of _ImageSchema class is disallowed.") -_ImageSchema.__init__ = _disallow_instance + + +_ImageUtils.__init__ = _disallow_instance def _test(): diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py index 95efa73f9e4c7..da835f7450698 100644 --- a/python/pyspark/ml/tests/test_image.py +++ b/python/pyspark/ml/tests/test_image.py @@ -18,7 +18,7 @@ import py4j -from pyspark.ml.image import ImageSchema +from pyspark.ml.image import ImageUtils from pyspark.testing.mlutils import PySparkTestCase, SparkSessionTestCase from pyspark.sql import HiveContext, Row from pyspark.testing.utils import QuietTest @@ -28,37 +28,40 @@ class ImageReaderTest(SparkSessionTestCase): def test_read_images(self): data_path = 'data/mllib/images/origin/kittens' - df = ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True) + df = self.spark.read.format("image") \ + .option("dropInvalid", True) \ + .option("recursiveFileLookup", True) \ + .load(data_path) self.assertEqual(df.count(), 4) first_row = df.take(1)[0][0] - array = ImageSchema.toNDArray(first_row) + array = ImageUtils.toNDArray(first_row) self.assertEqual(len(array), first_row[1]) - self.assertEqual(ImageSchema.toImage(array, origin=first_row[0]), first_row) - self.assertEqual(df.schema, ImageSchema.imageSchema) - self.assertEqual(df.schema["image"].dataType, ImageSchema.columnSchema) + self.assertEqual(ImageUtils.toImage(array, origin=first_row[0]), first_row) + self.assertEqual(df.schema, ImageUtils.imageSchema) + self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) expected = {'CV_8UC3': 16, 'Undefined': -1, 'CV_8U': 0, 'CV_8UC1': 0, 'CV_8UC4': 24} - self.assertEqual(ImageSchema.ocvTypes, expected) + self.assertEqual(ImageUtils.ocvTypes, expected) expected = ['origin', 'height', 'width', 'nChannels', 'mode', 'data'] - self.assertEqual(ImageSchema.imageFields, expected) - self.assertEqual(ImageSchema.undefinedImageType, "Undefined") + self.assertEqual(ImageUtils.imageFields, expected) + self.assertEqual(ImageUtils.undefinedImageType, "Undefined") with QuietTest(self.sc): self.assertRaisesRegexp( TypeError, "image argument should be pyspark.sql.types.Row; however", - lambda: ImageSchema.toNDArray("a")) + lambda: ImageUtils.toNDArray("a")) with QuietTest(self.sc): self.assertRaisesRegexp( ValueError, "image argument should have attributes specified in", - lambda: ImageSchema.toNDArray(Row(a=1))) + lambda: ImageUtils.toNDArray(Row(a=1))) with QuietTest(self.sc): self.assertRaisesRegexp( TypeError, "array argument should be numpy.ndarray; however, it got", - lambda: ImageSchema.toImage("a")) + lambda: ImageUtils.toImage("a")) class ImageReaderTest2(PySparkTestCase): @@ -95,8 +98,11 @@ def test_read_images_multiple_times(self): # This test case is to check if `ImageSchema.readImages` tries to # initiate Hive client multiple times. See SPARK-22651. data_path = 'data/mllib/images/origin/kittens' - ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True) - ImageSchema.readImages(data_path, recursive=True, dropImageFailures=True) + for _ in range(2): + self.spark.read.format("image") \ + .option("dropInvalid", True) \ + .option("recursiveFileLookup", True) \ + .load(data_path) if __name__ == "__main__": From 0a58e7021118e509c96b2dd264d9333a871e70ac Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 17:42:20 +0800 Subject: [PATCH 02/11] update --- .../ml/source/image/ImageFileFormat.scala | 43 +++++++++++++++++++ .../source/image/ImageFileFormatSuite.scala | 4 +- 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index 1a10c150896eb..b6c92f22a679d 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -129,4 +129,47 @@ object ImageFileFormat { private[image] def invalidImageRow(origin: String): Row = Row(Row(origin, -1, -1, -1, ImageUtils.ocvTypes(ImageUtils.undefinedImageType), Array.ofDim[Byte](0))) + + /** + * Gets the origin of the image + * + * @return The origin of the image + */ + def getOrigin(row: Row): String = row.getString(0) + + /** + * Gets the height of the image + * + * @return The height of the image + */ + def getHeight(row: Row): Int = row.getInt(1) + + /** + * Gets the width of the image + * + * @return The width of the image + */ + def getWidth(row: Row): Int = row.getInt(2) + + /** + * Gets the number of channels in the image + * + * @return The number of channels in the image + */ + def getNChannels(row: Row): Int = row.getInt(3) + + /** + * Gets the OpenCV representation as an int + * + * @return The OpenCV representation as an int + */ + def getMode(row: Row): Int = row.getInt(4) + + /** + * Gets the image data + * + * @return The image data + */ + def getData(row: Row): Array[Byte] = row.getAs[Array[Byte]](5) + } diff --git a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala index 3f94918d14832..296ec1c5cc942 100644 --- a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala @@ -21,12 +21,12 @@ import java.net.URI import java.nio.file.Paths import org.apache.spark.SparkFunSuite -import org.apache.spark.ml.source.image.ImageUtils._ import org.apache.spark.mllib.util.MLlibTestSparkContext import org.apache.spark.sql.Row import org.apache.spark.sql.functions.{col, substring_index} class ImageFileFormatSuite extends SparkFunSuite with MLlibTestSparkContext { + import ImageFileFormat._ // Single column of images named "image" private lazy val imagePath = "../data/mllib/images/partitioned" @@ -67,7 +67,7 @@ class ImageFileFormatSuite extends SparkFunSuite with MLlibTestSparkContext { assert(new URI(resultOrigin) === Paths.get(filePath).toAbsolutePath().normalize().toUri()) // Compare other columns in the row to be the same with the `invalidImageRow` - assert(result === ImageFileFormat.invalidImageRow(resultOrigin)) + assert(result === invalidImageRow(resultOrigin)) } test("image datasource partition test") { From 2bf66160c83fa187f3ed5e59aa8b0ada29713e61 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 20:34:37 +0800 Subject: [PATCH 03/11] update --- .../ml/source/image/ImageFileFormat.scala | 98 ++++++++++++++- .../spark/ml/source/image/ImageUtils.scala | 118 ------------------ python/pyspark/ml/image.py | 12 +- python/pyspark/ml/tests/test_image.py | 4 +- 4 files changed, 103 insertions(+), 129 deletions(-) delete mode 100644 mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index b6c92f22a679d..2b2224a3e2799 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -17,17 +17,22 @@ package org.apache.spark.ml.source.image +import java.awt.Color +import java.awt.color.ColorSpace +import java.io.ByteArrayInputStream +import javax.imageio.ImageIO + import com.google.common.io.{ByteStreams, Closeables} import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.mapreduce.Job +import scala.collection.JavaConverters._ import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.encoders.RowEncoder import org.apache.spark.sql.catalyst.expressions.UnsafeRow -import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, - PartitionedFile} +import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile} import org.apache.spark.sql.sources.{DataSourceRegister, Filter} import org.apache.spark.sql.types._ import org.apache.spark.util.SerializableConfiguration @@ -81,7 +86,7 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } finally { Closeables.close(stream, true) } - val resultOpt = ImageUtils.decode(origin, bytes) + val resultOpt = decode(origin, bytes) val filteredResult = if (imageSourceOptions.dropInvalid) { resultOpt.toIterator } else { @@ -100,6 +105,22 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } object ImageFileFormat { + + val undefinedImageType = "Undefined" + + /** + * (Scala-specific) OpenCV type mapping supported + */ + val ocvTypes: Map[String, Int] = Map( + undefinedImageType -> -1, + "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 + ) + + /** + * (Java-specific) OpenCV type mapping supported + */ + val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava + /** * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) */ @@ -127,7 +148,7 @@ object ImageFileFormat { * @return Row with the default values */ private[image] def invalidImageRow(origin: String): Row = - Row(Row(origin, -1, -1, -1, ImageUtils.ocvTypes(ImageUtils.undefinedImageType), + Row(Row(origin, -1, -1, -1, ocvTypes(undefinedImageType), Array.ofDim[Byte](0))) /** @@ -172,4 +193,73 @@ object ImageFileFormat { */ def getData(row: Row): Array[Byte] = row.getAs[Array[Byte]](5) + /** + * Convert the compressed image (jpeg, png, etc.) into OpenCV + * representation and store it in DataFrame Row + * + * @param origin Arbitrary string that identifies the image + * @param bytes Image bytes (for example, jpeg) + * @return DataFrame Row or None (if the decompression fails) + */ + def decode(origin: String, bytes: Array[Byte]): Option[Row] = { + + val img = try { + ImageIO.read(new ByteArrayInputStream(bytes)) + } catch { + // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. + // But do not catch the declared `IOException` (regarded as FileSystem failure) + case _: RuntimeException => null + } + + if (img == null) { + None + } else { + val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY + val hasAlpha = img.getColorModel.hasAlpha + + val height = img.getHeight + val width = img.getWidth + val (nChannels, mode) = if (isGray) { + (1, ocvTypes("CV_8UC1")) + } else if (hasAlpha) { + (4, ocvTypes("CV_8UC4")) + } else { + (3, ocvTypes("CV_8UC3")) + } + + val imageSize = height * width * nChannels + assert(imageSize < 1e9, "image is too large") + val decoded = Array.ofDim[Byte](imageSize) + + // Grayscale images in Java require special handling to get the correct intensity + if (isGray) { + var offset = 0 + val raster = img.getRaster + for (h <- 0 until height) { + for (w <- 0 until width) { + decoded(offset) = raster.getSample(w, h, 0).toByte + offset += 1 + } + } + } else { + var offset = 0 + for (h <- 0 until height) { + for (w <- 0 until width) { + val color = new Color(img.getRGB(w, h), hasAlpha) + decoded(offset) = color.getBlue.toByte + decoded(offset + 1) = color.getGreen.toByte + decoded(offset + 2) = color.getRed.toByte + if (hasAlpha) { + decoded(offset + 3) = color.getAlpha.toByte + } + offset += nChannels + } + } + } + + // the internal "Row" is needed, because the image is a single DataFrame column + Some(Row(Row(origin, height, width, nChannels, mode, decoded))) + } + } + } diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala deleted file mode 100644 index 95ae8f68a43da..0000000000000 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageUtils.scala +++ /dev/null @@ -1,118 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.spark.ml.source.image - -import java.awt.Color -import java.awt.color.ColorSpace -import java.io.ByteArrayInputStream -import javax.imageio.ImageIO - -import scala.collection.JavaConverters._ - -import org.apache.spark.sql.Row - -/** - * Defines the image util methods to read and manipulate images. - */ -private[spark] object ImageUtils { - - val undefinedImageType = "Undefined" - - /** - * (Scala-specific) OpenCV type mapping supported - */ - val ocvTypes: Map[String, Int] = Map( - undefinedImageType -> -1, - "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 - ) - - /** - * (Java-specific) OpenCV type mapping supported - */ - val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava - - /** - * Convert the compressed image (jpeg, png, etc.) into OpenCV - * representation and store it in DataFrame Row - * - * @param origin Arbitrary string that identifies the image - * @param bytes Image bytes (for example, jpeg) - * @return DataFrame Row or None (if the decompression fails) - */ - def decode(origin: String, bytes: Array[Byte]): Option[Row] = { - - val img = try { - ImageIO.read(new ByteArrayInputStream(bytes)) - } catch { - // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. - // But do not catch the declared `IOException` (regarded as FileSystem failure) - case _: RuntimeException => null - } - - if (img == null) { - None - } else { - val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY - val hasAlpha = img.getColorModel.hasAlpha - - val height = img.getHeight - val width = img.getWidth - val (nChannels, mode) = if (isGray) { - (1, ocvTypes("CV_8UC1")) - } else if (hasAlpha) { - (4, ocvTypes("CV_8UC4")) - } else { - (3, ocvTypes("CV_8UC3")) - } - - val imageSize = height * width * nChannels - assert(imageSize < 1e9, "image is too large") - val decoded = Array.ofDim[Byte](imageSize) - - // Grayscale images in Java require special handling to get the correct intensity - if (isGray) { - var offset = 0 - val raster = img.getRaster - for (h <- 0 until height) { - for (w <- 0 until width) { - decoded(offset) = raster.getSample(w, h, 0).toByte - offset += 1 - } - } - } else { - var offset = 0 - for (h <- 0 until height) { - for (w <- 0 until width) { - val color = new Color(img.getRGB(w, h), hasAlpha) - decoded(offset) = color.getBlue.toByte - decoded(offset + 1) = color.getGreen.toByte - decoded(offset + 2) = color.getRed.toByte - if (hasAlpha) { - decoded(offset + 3) = color.getAlpha.toByte - } - offset += nChannels - } - } - } - - // the internal "Row" is needed, because the image is a single DataFrame column - Some(Row(Row(origin, height, width, nChannels, mode, decoded))) - } - } - -} diff --git a/python/pyspark/ml/image.py b/python/pyspark/ml/image.py index 13e278e9a06e0..7d1a9d7e58a79 100644 --- a/python/pyspark/ml/image.py +++ b/python/pyspark/ml/image.py @@ -64,7 +64,7 @@ def imageSchema(self): if self._imageSchema is None: ctx = SparkContext._active_spark_context - jschema = ctx._jvm.org.apache.spark.ml.image.ImageSchema.imageSchema() + jschema = ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.schema() self._imageSchema = _parse_datatype_json_string(jschema.json()) return self._imageSchema @@ -80,7 +80,8 @@ def ocvTypes(self): if self._ocvTypes is None: ctx = SparkContext._active_spark_context - self._ocvTypes = dict(ctx._jvm.org.apache.spark.ml.image.ImageSchema.javaOcvTypes()) + self._ocvTypes = dict( + ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.javaOcvTypes()) return self._ocvTypes @property @@ -96,7 +97,7 @@ def columnSchema(self): if self._columnSchema is None: ctx = SparkContext._active_spark_context - jschema = ctx._jvm.org.apache.spark.ml.image.ImageSchema.columnSchema() + jschema = ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.columnSchema() self._columnSchema = _parse_datatype_json_string(jschema.json()) return self._columnSchema @@ -112,7 +113,8 @@ def imageFields(self): if self._imageFields is None: ctx = SparkContext._active_spark_context - self._imageFields = list(ctx._jvm.org.apache.spark.ml.image.ImageSchema.imageFields()) + self._imageFields = list( + ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.imageFields()) return self._imageFields @property @@ -126,7 +128,7 @@ def undefinedImageType(self): if self._undefinedImageType is None: ctx = SparkContext._active_spark_context self._undefinedImageType = \ - ctx._jvm.org.apache.spark.ml.image.ImageSchema.undefinedImageType() + ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.undefinedImageType() return self._undefinedImageType def toNDArray(self, image): diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py index da835f7450698..9b8263453bf20 100644 --- a/python/pyspark/ml/tests/test_image.py +++ b/python/pyspark/ml/tests/test_image.py @@ -37,8 +37,8 @@ def test_read_images(self): array = ImageUtils.toNDArray(first_row) self.assertEqual(len(array), first_row[1]) self.assertEqual(ImageUtils.toImage(array, origin=first_row[0]), first_row) - self.assertEqual(df.schema, ImageUtils.imageSchema) - self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) + # self.assertEqual(df.schema, ImageUtils.imageSchema) + # self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) expected = {'CV_8UC3': 16, 'Undefined': -1, 'CV_8U': 0, 'CV_8UC1': 0, 'CV_8UC4': 24} self.assertEqual(ImageUtils.ocvTypes, expected) expected = ['origin', 'height', 'width', 'nChannels', 'mode', 'data'] From a35e8b7dc3709a374da7b8791d03bbb4d72b2a9e Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 20:40:32 +0800 Subject: [PATCH 04/11] update --- python/pyspark/ml/image.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/pyspark/ml/image.py b/python/pyspark/ml/image.py index 7d1a9d7e58a79..4369fbd62271b 100644 --- a/python/pyspark/ml/image.py +++ b/python/pyspark/ml/image.py @@ -39,8 +39,8 @@ class _ImageUtils(object): """ - Internal class for `pyspark.ml.image.ImageSchema` attribute. Meant to be private and - not to be instantized. Use `pyspark.ml.image.ImageSchema` attribute to access the + Internal class for `pyspark.ml.image.ImageUtils` attribute. Meant to be private and + not to be instantized. Use `pyspark.ml.image.ImageUtils` attribute to access the APIs of this class. """ @@ -211,7 +211,7 @@ def toImage(self, array, origin=""): # Monkey patch to disallow instantiation of this class. def _disallow_instance(_): - raise RuntimeError("Creating instance of _ImageSchema class is disallowed.") + raise RuntimeError("Creating instance of _ImageUtils class is disallowed.") _ImageUtils.__init__ = _disallow_instance From dd711677e14fd9bfd81f75cad8bc70b16419c24a Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 22:01:10 +0800 Subject: [PATCH 05/11] update --- .../apache/spark/ml/source/image/ImageFileFormat.scala | 10 +++++----- python/pyspark/ml/tests/test_image.py | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index 2b2224a3e2799..c01f10ff3139c 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -126,13 +126,13 @@ object ImageFileFormat { */ private[image] val columnSchema = StructType( StructField("origin", StringType, true) :: - StructField("height", IntegerType, false) :: - StructField("width", IntegerType, false) :: - StructField("nChannels", IntegerType, false) :: + StructField("height", IntegerType, true) :: + StructField("width", IntegerType, true) :: + StructField("nChannels", IntegerType, true) :: // OpenCV-compatible type: CV_8UC3 in most cases - StructField("mode", IntegerType, false) :: + StructField("mode", IntegerType, true) :: // Bytes in OpenCV-compatible order: row-wise BGR in most cases - StructField("data", BinaryType, false) :: Nil) + StructField("data", BinaryType, true) :: Nil) private[image] val imageFields: Array[String] = columnSchema.fieldNames diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py index 9b8263453bf20..da835f7450698 100644 --- a/python/pyspark/ml/tests/test_image.py +++ b/python/pyspark/ml/tests/test_image.py @@ -37,8 +37,8 @@ def test_read_images(self): array = ImageUtils.toNDArray(first_row) self.assertEqual(len(array), first_row[1]) self.assertEqual(ImageUtils.toImage(array, origin=first_row[0]), first_row) - # self.assertEqual(df.schema, ImageUtils.imageSchema) - # self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) + self.assertEqual(df.schema, ImageUtils.imageSchema) + self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) expected = {'CV_8UC3': 16, 'Undefined': -1, 'CV_8U': 0, 'CV_8UC1': 0, 'CV_8UC4': 24} self.assertEqual(ImageUtils.ocvTypes, expected) expected = ['origin', 'height', 'width', 'nChannels', 'mode', 'data'] From 07b29395c8cc6f47cfc9d3177bd35d0d659c5f60 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 24 Jul 2019 22:34:14 +0800 Subject: [PATCH 06/11] address comments --- .../org/apache/spark/ml/source/image/ImageFileFormat.scala | 5 +++-- python/pyspark/ml/image.py | 6 +++--- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index c01f10ff3139c..fc52d9f556a81 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -22,11 +22,12 @@ import java.awt.color.ColorSpace import java.io.ByteArrayInputStream import javax.imageio.ImageIO +import scala.collection.JavaConverters._ + import com.google.common.io.{ByteStreams, Closeables} import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.mapreduce.Job -import scala.collection.JavaConverters._ import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.InternalRow @@ -34,7 +35,7 @@ import org.apache.spark.sql.catalyst.encoders.RowEncoder import org.apache.spark.sql.catalyst.expressions.UnsafeRow import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile} import org.apache.spark.sql.sources.{DataSourceRegister, Filter} -import org.apache.spark.sql.types._ +import org.apache.spark.sql.types.{BinaryType, IntegerType, StringType, StructField, StructType} import org.apache.spark.util.SerializableConfiguration private[image] class ImageFileFormat extends FileFormat with DataSourceRegister { diff --git a/python/pyspark/ml/image.py b/python/pyspark/ml/image.py index 4369fbd62271b..9cad3006beb91 100644 --- a/python/pyspark/ml/image.py +++ b/python/pyspark/ml/image.py @@ -16,11 +16,11 @@ # """ -.. attribute:: ImageSchema +.. attribute:: ImageUtils - An attribute of this module that contains the instance of :class:`_ImageSchema`. + An attribute of this module that contains the instance of :class:`_ImageUtils`. -.. autoclass:: _ImageSchema +.. autoclass:: _ImageUtils :members: """ From fb50ec5524ab9b91ddae5f26e89621d531206a25 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Thu, 25 Jul 2019 00:45:45 +0800 Subject: [PATCH 07/11] update mima excludes --- project/MimaExcludes.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 51d5861eebef6..379c271e8789e 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -380,7 +380,11 @@ object MimaExcludes { // [SPARK-28556][SQL] QueryExecutionListener should also notify Error ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.sql.util.QueryExecutionListener.onFailure"), - ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.util.QueryExecutionListener.onFailure") + ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.util.QueryExecutionListener.onFailure"), + + // [SPARK-25382][SQL][PYSPARK] Remove ImageSchema.readImages in 3.0 + ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ml.image.ImageSchema"), + ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ml.image.ImageSchema$") ) // Exclude rules for 2.4.x From 747922c3a56eaf814f968088c1f1447fd951e9e3 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Thu, 25 Jul 2019 09:46:49 +0800 Subject: [PATCH 08/11] fix py test --- .../org/apache/spark/ml/source/image/ImageFileFormat.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index fc52d9f556a81..08a9b32a35307 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -125,7 +125,7 @@ object ImageFileFormat { /** * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) */ - private[image] val columnSchema = StructType( + val columnSchema = StructType( StructField("origin", StringType, true) :: StructField("height", IntegerType, true) :: StructField("width", IntegerType, true) :: @@ -135,7 +135,7 @@ object ImageFileFormat { // Bytes in OpenCV-compatible order: row-wise BGR in most cases StructField("data", BinaryType, true) :: Nil) - private[image] val imageFields: Array[String] = columnSchema.fieldNames + val imageFields: Array[String] = columnSchema.fieldNames /** * DataFrame with a single column of images named "image" (nullable) From 4acb544f5e9793d1a90dc87174cc4a697393a1df Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Tue, 30 Jul 2019 20:01:08 +0800 Subject: [PATCH 09/11] update --- .../apache/spark/ml/image/ImageSchema.scala | 194 ++++++++++++++++++ .../ml/source/image/ImageFileFormat.scala | 184 +---------------- .../source/image/ImageFileFormatSuite.scala | 20 +- python/pyspark/ml/image.py | 36 ++-- python/pyspark/ml/tests/test_image.py | 32 ++- 5 files changed, 253 insertions(+), 213 deletions(-) create mode 100644 mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala diff --git a/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala b/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala new file mode 100644 index 0000000000000..03136261dd5c9 --- /dev/null +++ b/mllib/src/main/scala/org/apache/spark/ml/image/ImageSchema.scala @@ -0,0 +1,194 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.spark.ml.image + +import java.awt.Color +import java.awt.color.ColorSpace +import java.io.ByteArrayInputStream +import javax.imageio.ImageIO + +import scala.collection.JavaConverters._ + +import org.apache.spark.annotation.{Experimental, Since} +import org.apache.spark.input.PortableDataStream +import org.apache.spark.sql.{DataFrame, Row, SparkSession} +import org.apache.spark.sql.types._ + +/** + * :: Experimental :: + * Defines the image schema and methods to read and manipulate images. + */ +@Experimental +@Since("2.3.0") +object ImageSchema { + + val undefinedImageType = "Undefined" + + /** + * (Scala-specific) OpenCV type mapping supported + */ + val ocvTypes: Map[String, Int] = Map( + undefinedImageType -> -1, + "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 + ) + + /** + * (Java-specific) OpenCV type mapping supported + */ + val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava + + /** + * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) + */ + val columnSchema = StructType( + StructField("origin", StringType, true) :: + StructField("height", IntegerType, false) :: + StructField("width", IntegerType, false) :: + StructField("nChannels", IntegerType, false) :: + // OpenCV-compatible type: CV_8UC3 in most cases + StructField("mode", IntegerType, false) :: + // Bytes in OpenCV-compatible order: row-wise BGR in most cases + StructField("data", BinaryType, false) :: Nil) + + val imageFields: Array[String] = columnSchema.fieldNames + + /** + * DataFrame with a single column of images named "image" (nullable) + */ + val imageSchema = StructType(StructField("image", columnSchema, true) :: Nil) + + /** + * Gets the origin of the image + * + * @return The origin of the image + */ + def getOrigin(row: Row): String = row.getString(0) + + /** + * Gets the height of the image + * + * @return The height of the image + */ + def getHeight(row: Row): Int = row.getInt(1) + + /** + * Gets the width of the image + * + * @return The width of the image + */ + def getWidth(row: Row): Int = row.getInt(2) + + /** + * Gets the number of channels in the image + * + * @return The number of channels in the image + */ + def getNChannels(row: Row): Int = row.getInt(3) + + /** + * Gets the OpenCV representation as an int + * + * @return The OpenCV representation as an int + */ + def getMode(row: Row): Int = row.getInt(4) + + /** + * Gets the image data + * + * @return The image data + */ + def getData(row: Row): Array[Byte] = row.getAs[Array[Byte]](5) + + /** + * Default values for the invalid image + * + * @param origin Origin of the invalid image + * @return Row with the default values + */ + private[spark] def invalidImageRow(origin: String): Row = + Row(Row(origin, -1, -1, -1, ocvTypes(undefinedImageType), Array.ofDim[Byte](0))) + + /** + * Convert the compressed image (jpeg, png, etc.) into OpenCV + * representation and store it in DataFrame Row + * + * @param origin Arbitrary string that identifies the image + * @param bytes Image bytes (for example, jpeg) + * @return DataFrame Row or None (if the decompression fails) + */ + private[spark] def decode(origin: String, bytes: Array[Byte]): Option[Row] = { + + val img = try { + ImageIO.read(new ByteArrayInputStream(bytes)) + } catch { + // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. + // But do not catch the declared `IOException` (regarded as FileSystem failure) + case _: RuntimeException => null + } + + if (img == null) { + None + } else { + val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY + val hasAlpha = img.getColorModel.hasAlpha + + val height = img.getHeight + val width = img.getWidth + val (nChannels, mode) = if (isGray) { + (1, ocvTypes("CV_8UC1")) + } else if (hasAlpha) { + (4, ocvTypes("CV_8UC4")) + } else { + (3, ocvTypes("CV_8UC3")) + } + + val imageSize = height * width * nChannels + assert(imageSize < 1e9, "image is too large") + val decoded = Array.ofDim[Byte](imageSize) + + // Grayscale images in Java require special handling to get the correct intensity + if (isGray) { + var offset = 0 + val raster = img.getRaster + for (h <- 0 until height) { + for (w <- 0 until width) { + decoded(offset) = raster.getSample(w, h, 0).toByte + offset += 1 + } + } + } else { + var offset = 0 + for (h <- 0 until height) { + for (w <- 0 until width) { + val color = new Color(img.getRGB(w, h), hasAlpha) + decoded(offset) = color.getBlue.toByte + decoded(offset + 1) = color.getGreen.toByte + decoded(offset + 2) = color.getRed.toByte + if (hasAlpha) { + decoded(offset + 3) = color.getAlpha.toByte + } + offset += nChannels + } + } + } + + // the internal "Row" is needed, because the image is a single DataFrame column + Some(Row(Row(origin, height, width, nChannels, mode, decoded))) + } + } +} diff --git a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala index 08a9b32a35307..c3321447e3c96 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/source/image/ImageFileFormat.scala @@ -17,34 +17,28 @@ package org.apache.spark.ml.source.image -import java.awt.Color -import java.awt.color.ColorSpace -import java.io.ByteArrayInputStream -import javax.imageio.ImageIO - -import scala.collection.JavaConverters._ - import com.google.common.io.{ByteStreams, Closeables} import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.mapreduce.Job -import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.ml.image.ImageSchema +import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.encoders.RowEncoder -import org.apache.spark.sql.catalyst.expressions.UnsafeRow -import org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriterFactory, PartitionedFile} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, UnsafeRow} +import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap +import org.apache.spark.sql.execution.datasources.{DataSource, FileFormat, OutputWriterFactory, PartitionedFile} import org.apache.spark.sql.sources.{DataSourceRegister, Filter} -import org.apache.spark.sql.types.{BinaryType, IntegerType, StringType, StructField, StructType} +import org.apache.spark.sql.types.StructType import org.apache.spark.util.SerializableConfiguration private[image] class ImageFileFormat extends FileFormat with DataSourceRegister { - import ImageFileFormat._ override def inferSchema( sparkSession: SparkSession, options: Map[String, String], - files: Seq[FileStatus]): Option[StructType] = Some(schema) + files: Seq[FileStatus]): Option[StructType] = Some(ImageSchema.imageSchema) override def prepareWrite( sparkSession: SparkSession, @@ -87,11 +81,11 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } finally { Closeables.close(stream, true) } - val resultOpt = decode(origin, bytes) + val resultOpt = ImageSchema.decode(origin, bytes) val filteredResult = if (imageSourceOptions.dropInvalid) { resultOpt.toIterator } else { - Iterator(resultOpt.getOrElse(invalidImageRow(origin))) + Iterator(resultOpt.getOrElse(ImageSchema.invalidImageRow(origin))) } if (requiredSchema.isEmpty) { @@ -104,163 +98,3 @@ private[image] class ImageFileFormat extends FileFormat with DataSourceRegister } } } - -object ImageFileFormat { - - val undefinedImageType = "Undefined" - - /** - * (Scala-specific) OpenCV type mapping supported - */ - val ocvTypes: Map[String, Int] = Map( - undefinedImageType -> -1, - "CV_8U" -> 0, "CV_8UC1" -> 0, "CV_8UC3" -> 16, "CV_8UC4" -> 24 - ) - - /** - * (Java-specific) OpenCV type mapping supported - */ - val javaOcvTypes: java.util.Map[String, Int] = ocvTypes.asJava - - /** - * Schema for the image column: Row(String, Int, Int, Int, Int, Array[Byte]) - */ - val columnSchema = StructType( - StructField("origin", StringType, true) :: - StructField("height", IntegerType, true) :: - StructField("width", IntegerType, true) :: - StructField("nChannels", IntegerType, true) :: - // OpenCV-compatible type: CV_8UC3 in most cases - StructField("mode", IntegerType, true) :: - // Bytes in OpenCV-compatible order: row-wise BGR in most cases - StructField("data", BinaryType, true) :: Nil) - - val imageFields: Array[String] = columnSchema.fieldNames - - /** - * DataFrame with a single column of images named "image" (nullable) - */ - val schema = StructType(StructField("image", columnSchema, true) :: Nil) - - /** - * Default values for the invalid image - * - * @param origin Origin of the invalid image - * @return Row with the default values - */ - private[image] def invalidImageRow(origin: String): Row = - Row(Row(origin, -1, -1, -1, ocvTypes(undefinedImageType), - Array.ofDim[Byte](0))) - - /** - * Gets the origin of the image - * - * @return The origin of the image - */ - def getOrigin(row: Row): String = row.getString(0) - - /** - * Gets the height of the image - * - * @return The height of the image - */ - def getHeight(row: Row): Int = row.getInt(1) - - /** - * Gets the width of the image - * - * @return The width of the image - */ - def getWidth(row: Row): Int = row.getInt(2) - - /** - * Gets the number of channels in the image - * - * @return The number of channels in the image - */ - def getNChannels(row: Row): Int = row.getInt(3) - - /** - * Gets the OpenCV representation as an int - * - * @return The OpenCV representation as an int - */ - def getMode(row: Row): Int = row.getInt(4) - - /** - * Gets the image data - * - * @return The image data - */ - def getData(row: Row): Array[Byte] = row.getAs[Array[Byte]](5) - - /** - * Convert the compressed image (jpeg, png, etc.) into OpenCV - * representation and store it in DataFrame Row - * - * @param origin Arbitrary string that identifies the image - * @param bytes Image bytes (for example, jpeg) - * @return DataFrame Row or None (if the decompression fails) - */ - def decode(origin: String, bytes: Array[Byte]): Option[Row] = { - - val img = try { - ImageIO.read(new ByteArrayInputStream(bytes)) - } catch { - // Catch runtime exception because `ImageIO` may throw unexcepted `RuntimeException`. - // But do not catch the declared `IOException` (regarded as FileSystem failure) - case _: RuntimeException => null - } - - if (img == null) { - None - } else { - val isGray = img.getColorModel.getColorSpace.getType == ColorSpace.TYPE_GRAY - val hasAlpha = img.getColorModel.hasAlpha - - val height = img.getHeight - val width = img.getWidth - val (nChannels, mode) = if (isGray) { - (1, ocvTypes("CV_8UC1")) - } else if (hasAlpha) { - (4, ocvTypes("CV_8UC4")) - } else { - (3, ocvTypes("CV_8UC3")) - } - - val imageSize = height * width * nChannels - assert(imageSize < 1e9, "image is too large") - val decoded = Array.ofDim[Byte](imageSize) - - // Grayscale images in Java require special handling to get the correct intensity - if (isGray) { - var offset = 0 - val raster = img.getRaster - for (h <- 0 until height) { - for (w <- 0 until width) { - decoded(offset) = raster.getSample(w, h, 0).toByte - offset += 1 - } - } - } else { - var offset = 0 - for (h <- 0 until height) { - for (w <- 0 until width) { - val color = new Color(img.getRGB(w, h), hasAlpha) - decoded(offset) = color.getBlue.toByte - decoded(offset + 1) = color.getGreen.toByte - decoded(offset + 2) = color.getRed.toByte - if (hasAlpha) { - decoded(offset + 3) = color.getAlpha.toByte - } - offset += nChannels - } - } - } - - // the internal "Row" is needed, because the image is a single DataFrame column - Some(Row(Row(origin, height, width, nChannels, mode, decoded))) - } - } - -} diff --git a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala index 296ec1c5cc942..0ec2747be6585 100644 --- a/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/ml/source/image/ImageFileFormatSuite.scala @@ -21,17 +21,35 @@ import java.net.URI import java.nio.file.Paths import org.apache.spark.SparkFunSuite +import org.apache.spark.ml.image.ImageSchema._ import org.apache.spark.mllib.util.MLlibTestSparkContext import org.apache.spark.sql.Row import org.apache.spark.sql.functions.{col, substring_index} class ImageFileFormatSuite extends SparkFunSuite with MLlibTestSparkContext { - import ImageFileFormat._ // Single column of images named "image" private lazy val imagePath = "../data/mllib/images/partitioned" private lazy val recursiveImagePath = "../data/mllib/images" + test("Smoke test: create basic ImageSchema dataframe") { + val origin = "path" + val width = 1 + val height = 1 + val nChannels = 3 + val data = Array[Byte](0, 0, 0) + val mode = ocvTypes("CV_8UC3") + + // Internal Row corresponds to image StructType + val rows = Seq(Row(Row(origin, height, width, nChannels, mode, data)), + Row(Row(null, height, width, nChannels, mode, data))) + val rdd = sc.makeRDD(rows) + val df = spark.createDataFrame(rdd, imageSchema) + + assert(df.count === 2, "incorrect image count") + assert(df.schema("image").dataType == columnSchema, "data do not fit ImageSchema") + } + test("image datasource count test") { val df1 = spark.read.format("image").load(imagePath) assert(df1.count === 9) diff --git a/python/pyspark/ml/image.py b/python/pyspark/ml/image.py index 9cad3006beb91..4fb1036fbab89 100644 --- a/python/pyspark/ml/image.py +++ b/python/pyspark/ml/image.py @@ -16,11 +16,11 @@ # """ -.. attribute:: ImageUtils +.. attribute:: ImageSchema - An attribute of this module that contains the instance of :class:`_ImageUtils`. + An attribute of this module that contains the instance of :class:`_ImageSchema`. -.. autoclass:: _ImageUtils +.. autoclass:: _ImageSchema :members: """ @@ -34,13 +34,13 @@ from pyspark.sql.types import Row, _create_row, _parse_datatype_json_string from pyspark.sql import DataFrame, SparkSession -__all__ = ["ImageUtils"] +__all__ = ["ImageSchema"] -class _ImageUtils(object): +class _ImageSchema(object): """ - Internal class for `pyspark.ml.image.ImageUtils` attribute. Meant to be private and - not to be instantized. Use `pyspark.ml.image.ImageUtils` attribute to access the + Internal class for `pyspark.ml.image.ImageSchema` attribute. Meant to be private and + not to be instantized. Use `pyspark.ml.image.ImageSchema` attribute to access the APIs of this class. """ @@ -64,7 +64,7 @@ def imageSchema(self): if self._imageSchema is None: ctx = SparkContext._active_spark_context - jschema = ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.schema() + jschema = ctx._jvm.org.apache.spark.ml.image.ImageSchema.imageSchema() self._imageSchema = _parse_datatype_json_string(jschema.json()) return self._imageSchema @@ -80,8 +80,7 @@ def ocvTypes(self): if self._ocvTypes is None: ctx = SparkContext._active_spark_context - self._ocvTypes = dict( - ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.javaOcvTypes()) + self._ocvTypes = dict(ctx._jvm.org.apache.spark.ml.image.ImageSchema.javaOcvTypes()) return self._ocvTypes @property @@ -97,7 +96,7 @@ def columnSchema(self): if self._columnSchema is None: ctx = SparkContext._active_spark_context - jschema = ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.columnSchema() + jschema = ctx._jvm.org.apache.spark.ml.image.ImageSchema.columnSchema() self._columnSchema = _parse_datatype_json_string(jschema.json()) return self._columnSchema @@ -113,8 +112,7 @@ def imageFields(self): if self._imageFields is None: ctx = SparkContext._active_spark_context - self._imageFields = list( - ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.imageFields()) + self._imageFields = list(ctx._jvm.org.apache.spark.ml.image.ImageSchema.imageFields()) return self._imageFields @property @@ -128,7 +126,7 @@ def undefinedImageType(self): if self._undefinedImageType is None: ctx = SparkContext._active_spark_context self._undefinedImageType = \ - ctx._jvm.org.apache.spark.ml.source.image.ImageFileFormat.undefinedImageType() + ctx._jvm.org.apache.spark.ml.image.ImageSchema.undefinedImageType() return self._undefinedImageType def toNDArray(self, image): @@ -180,7 +178,7 @@ def toImage(self, array, origin=""): raise ValueError("Invalid array shape") height, width, nChannels = array.shape - ocvTypes = ImageUtils.ocvTypes + ocvTypes = ImageSchema.ocvTypes if nChannels == 1: mode = ocvTypes["CV_8UC1"] elif nChannels == 3: @@ -206,15 +204,13 @@ def toImage(self, array, origin=""): [origin, height, width, nChannels, mode, data]) -ImageUtils = _ImageUtils() +ImageSchema = _ImageSchema() # Monkey patch to disallow instantiation of this class. def _disallow_instance(_): - raise RuntimeError("Creating instance of _ImageUtils class is disallowed.") - - -_ImageUtils.__init__ = _disallow_instance + raise RuntimeError("Creating instance of _ImageSchema class is disallowed.") +_ImageSchema.__init__ = _disallow_instance def _test(): diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py index da835f7450698..bc0b354e2da0f 100644 --- a/python/pyspark/ml/tests/test_image.py +++ b/python/pyspark/ml/tests/test_image.py @@ -18,13 +18,13 @@ import py4j -from pyspark.ml.image import ImageUtils +from pyspark.ml.image import ImageSchema from pyspark.testing.mlutils import PySparkTestCase, SparkSessionTestCase from pyspark.sql import HiveContext, Row from pyspark.testing.utils import QuietTest -class ImageReaderTest(SparkSessionTestCase): +class ImageFileFormatTest(SparkSessionTestCase): def test_read_images(self): data_path = 'data/mllib/images/origin/kittens' @@ -34,41 +34,39 @@ def test_read_images(self): .load(data_path) self.assertEqual(df.count(), 4) first_row = df.take(1)[0][0] - array = ImageUtils.toNDArray(first_row) + array = ImageSchema.toNDArray(first_row) self.assertEqual(len(array), first_row[1]) - self.assertEqual(ImageUtils.toImage(array, origin=first_row[0]), first_row) - self.assertEqual(df.schema, ImageUtils.imageSchema) - self.assertEqual(df.schema["image"].dataType, ImageUtils.columnSchema) + self.assertEqual(ImageSchema.toImage(array, origin=first_row[0]), first_row) expected = {'CV_8UC3': 16, 'Undefined': -1, 'CV_8U': 0, 'CV_8UC1': 0, 'CV_8UC4': 24} - self.assertEqual(ImageUtils.ocvTypes, expected) + self.assertEqual(ImageSchema.ocvTypes, expected) expected = ['origin', 'height', 'width', 'nChannels', 'mode', 'data'] - self.assertEqual(ImageUtils.imageFields, expected) - self.assertEqual(ImageUtils.undefinedImageType, "Undefined") + self.assertEqual(ImageSchema.imageFields, expected) + self.assertEqual(ImageSchema.undefinedImageType, "Undefined") with QuietTest(self.sc): self.assertRaisesRegexp( TypeError, "image argument should be pyspark.sql.types.Row; however", - lambda: ImageUtils.toNDArray("a")) + lambda: ImageSchema.toNDArray("a")) with QuietTest(self.sc): self.assertRaisesRegexp( ValueError, "image argument should have attributes specified in", - lambda: ImageUtils.toNDArray(Row(a=1))) + lambda: ImageSchema.toNDArray(Row(a=1))) with QuietTest(self.sc): self.assertRaisesRegexp( TypeError, "array argument should be numpy.ndarray; however, it got", - lambda: ImageUtils.toImage("a")) + lambda: ImageSchema.toImage("a")) -class ImageReaderTest2(PySparkTestCase): +class ImageFileFormatOnHiveContextTest(PySparkTestCase): @classmethod def setUpClass(cls): - super(ImageReaderTest2, cls).setUpClass() + super(ImageFileFormatOnHiveContextTest, cls).setUpClass() cls.hive_available = True # Note that here we enable Hive's support. cls.spark = None @@ -89,16 +87,16 @@ def setUp(self): @classmethod def tearDownClass(cls): - super(ImageReaderTest2, cls).tearDownClass() + super(ImageFileFormatOnHiveContextTest, cls).tearDownClass() if cls.spark is not None: cls.spark.sparkSession.stop() cls.spark = None def test_read_images_multiple_times(self): - # This test case is to check if `ImageSchema.readImages` tries to + # This test case is to check if ImageFileFormat tries to # initiate Hive client multiple times. See SPARK-22651. data_path = 'data/mllib/images/origin/kittens' - for _ in range(2): + for i in range(2): self.spark.read.format("image") \ .option("dropInvalid", True) \ .option("recursiveFileLookup", True) \ From 02b4853ba4d8babfd86a497162989bac18b987d2 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Tue, 30 Jul 2019 20:24:57 +0800 Subject: [PATCH 10/11] update mima excludes --- project/MimaExcludes.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 379c271e8789e..4afca5ac82e86 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -383,8 +383,8 @@ object MimaExcludes { ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.util.QueryExecutionListener.onFailure"), // [SPARK-25382][SQL][PYSPARK] Remove ImageSchema.readImages in 3.0 - ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ml.image.ImageSchema"), - ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.ml.image.ImageSchema$") + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.ml.image.ImageSchema.readImages"), + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.ml.image.ImageSchema.readImages") ) // Exclude rules for 2.4.x From 6460c7a33d2ff9c61e63117ac6362469a834a9e3 Mon Sep 17 00:00:00 2001 From: WeichenXu Date: Wed, 31 Jul 2019 09:48:07 +0800 Subject: [PATCH 11/11] address comments --- python/pyspark/ml/tests/test_image.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/python/pyspark/ml/tests/test_image.py b/python/pyspark/ml/tests/test_image.py index bc0b354e2da0f..0008b0b670d34 100644 --- a/python/pyspark/ml/tests/test_image.py +++ b/python/pyspark/ml/tests/test_image.py @@ -34,6 +34,11 @@ def test_read_images(self): .load(data_path) self.assertEqual(df.count(), 4) first_row = df.take(1)[0][0] + # compare `schema.simpleString()` instead of directly compare schema, + # because the df loaded from datasouce may change schema column nullability. + self.assertEqual(df.schema.simpleString(), ImageSchema.imageSchema.simpleString()) + self.assertEqual(df.schema["image"].dataType.simpleString(), + ImageSchema.columnSchema.simpleString()) array = ImageSchema.toNDArray(first_row) self.assertEqual(len(array), first_row[1]) self.assertEqual(ImageSchema.toImage(array, origin=first_row[0]), first_row)