Skip to content

Commit

Permalink
Merge branch 'streaming-default' of https://github.com/svotaw/SynapseML…
Browse files Browse the repository at this point in the history 
… into streaming-default
  • Loading branch information
@svotaw
svotaw committed Oct 16, 2023
2 parents 212164b + b5556f9 commit d0690a3
Show file tree
Hide file tree
Showing 11 changed files with 353 additions and 97 deletions.
48 changes: 48 additions & 0 deletions cognitive/src/main/scala/com/microsoft/azure/synapse/ml/cognitive/text/TextAnalytics.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -270,8 +270,14 @@ trait HasStringIndexType extends HasServiceParams {
def setStringIndexType(v: String): this.type = setScalarParam(stringIndexType, v)
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object TextSentiment extends ComplexParamsReadable[TextSentiment]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class TextSentiment(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand Down Expand Up @@ -302,8 +308,14 @@ class TextSentiment(override val uid: String)

}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object KeyPhraseExtractor extends ComplexParamsReadable[KeyPhraseExtractor]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class KeyPhraseExtractor(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand All @@ -319,8 +331,14 @@ class KeyPhraseExtractor(override val uid: String)
override def urlPath: String = "/text/analytics/v3.1/keyPhrases"
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object NER extends ComplexParamsReadable[NER]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class NER(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand All @@ -336,8 +354,14 @@ class NER(override val uid: String)
override def urlPath: String = "/text/analytics/v3.1/entities/recognition/general"
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object PII extends ComplexParamsReadable[PII]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class PII(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand All @@ -363,8 +387,14 @@ class PII(override val uid: String)
override def urlPath: String = "/text/analytics/v3.1/entities/recognition/pii"
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object LanguageDetector extends ComplexParamsReadable[LanguageDetector]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class LanguageDetector(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand All @@ -380,8 +410,14 @@ class LanguageDetector(override val uid: String)
override def urlPath: String = "/text/analytics/v3.1/languages"
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object EntityDetector extends ComplexParamsReadable[EntityDetector]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class EntityDetector(override val uid: String)
extends TextAnalyticsBase(uid) with HasStringIndexType with HasHandler {
logClass()
Expand All @@ -397,8 +433,14 @@ class EntityDetector(override val uid: String)
override def urlPath: String = "/text/analytics/v3.1/entities/linking"
}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object AnalyzeHealthText extends ComplexParamsReadable[AnalyzeHealthText]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class AnalyzeHealthText(override val uid: String)
extends TextAnalyticsBaseNoBinding(uid)
with HasUnpackedBinding
Expand Down Expand Up @@ -441,8 +483,14 @@ class AnalyzeHealthText(override val uid: String)

}

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
object TextAnalyze extends ComplexParamsReadable[TextAnalyze]

@deprecated("This is an older version of the text analytics cognitive service" +
" and will be removed in v1.0.0 please use" +
" com.microsoft.azure.synapse.ml.cognitive.language.AnalyzeText instead", "v0.11.3")
class TextAnalyze(override val uid: String) extends TextAnalyticsBaseNoBinding(uid)
with BasicAsyncReply {
logClass()
Expand Down
6 changes: 2 additions & 4 deletions cognitive/src/test/scala/com/microsoft/azure/synapse/ml/cognitive/face/FaceSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,7 @@ class DetectFaceSuite extends TransformerFuzzing[DetectFace] with CognitiveKey {
.setOutputCol("face")
.setReturnFaceId(true)
.setReturnFaceLandmarks(true)
.setReturnFaceAttributes(Seq(
"age", "gender", "headPose", "smile", "facialHair", "glasses", "emotion",
"hair", "makeup", "occlusion", "accessories", "blur", "exposure", "noise"))
.setReturnFaceAttributes(Seq("exposure"))

override def assertDFEq(df1: DataFrame, df2: DataFrame)(implicit eq: Equality[DataFrame]): Unit = {
def prep(df: DataFrame) = df.select(explode(col("face"))).select("col.*").drop("faceId")
Expand All @@ -45,7 +43,7 @@ class DetectFaceSuite extends TransformerFuzzing[DetectFace] with CognitiveKey {
val fromRow = Face.makeFromRowConverter

val f1 = fromRow(results.select("face").collect().head.getSeq[Row](0).head)
assert(f1.faceAttributes.get.age.get > 20)
assert(f1.faceAttributes.get.exposure.get.value != 0.0)

results.show(truncate = false)
}
Expand Down
156 changes: 78 additions & 78 deletions ...ive/src/test/scala/com/microsoft/azure/synapse/ml/cognitive/text/TextAnalyticsSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -291,81 +291,81 @@ class AnalyzeHealthTextSuite extends TATestBase[AnalyzeHealthText] {

}

class TextAnalyzeSuite extends TransformerFuzzing[TextAnalyze] with TextEndpoint {

import spark.implicits._

implicit val doubleEquality: Equality[Double] = TolerantNumerics.tolerantDoubleEquality(1e-3)

def df: DataFrame = Seq(
("en", "I had a wonderful trip to Seattle last week and visited Microsoft."),
("en", "Another document bites the dust"),
(null, "ich bin ein berliner"),
(null, null),
("en", null),
("invalid", "This is irrelevant as the language is invalid")
).toDF("language", "text")

def model: TextAnalyze = new TextAnalyze()
.setSubscriptionKey(textKey)
.setLocation(textApiLocation)
.setLanguageCol("language")
.setOutputCol("response")
.setErrorCol("error")

def prepResults(df: DataFrame): Seq[Option[UnpackedTextAnalyzeResponse]] = {
val fromRow = model.unpackedResponseBinding.makeFromRowConverter
df.collect().map(row => Option(row.getAs[Row](model.getOutputCol)).map(fromRow)).toList
}

test("Basic usage") {
val topResult = prepResults(model.transform(df.coalesce(1))).head.get
assert(topResult.pii.get.document.get.entities.head.text == "last week")
assert(topResult.sentimentAnalysis.get.document.get.sentiment == "positive")
assert(topResult.entityLinking.get.document.get.entities.head.dataSource == "Wikipedia")
assert(topResult.keyPhraseExtraction.get.document.get.keyPhrases.head == "wonderful trip")
assert(topResult.entityRecognition.get.document.get.entities.head.text == "trip")
}

test("Manual Batching") {
val batchedDf = new FixedMiniBatchTransformer().setBatchSize(10).transform(df.coalesce(1))
val resultDF = new FlattenBatch().transform(model.transform(batchedDf))
val topResult = prepResults(resultDF).head.get
assert(topResult.pii.get.document.get.entities.head.text == "last week")
assert(topResult.sentimentAnalysis.get.document.get.sentiment == "positive")
assert(topResult.entityLinking.get.document.get.entities.head.dataSource == "Wikipedia")
assert(topResult.keyPhraseExtraction.get.document.get.keyPhrases.head == "wonderful trip")
assert(topResult.entityRecognition.get.document.get.entities.head.text == "trip")
}

test("Large Batching") {
val bigDF = (0 until 25).map(i => s"This is fantastic sentence number $i").toDF("text")
val model2 = model.setLanguage("en").setBatchSize(25)
val results = prepResults(model2.transform(bigDF.coalesce(1)))
assert(results.length == 25)
assert(results(24).get.sentimentAnalysis.get.document.get.sentiment == "positive")
}

test("Exceeded Retries Info") {
val badModel = model
.setPollingDelay(0)
.setInitialPollingDelay(0)
.setMaxPollingRetries(1)

val results = badModel
.setSuppressMaxRetriesException(true)
.transform(df.coalesce(1))
assert(results.where(!col("error").isNull).count() > 0)

assertThrows[SparkException] {
badModel.setSuppressMaxRetriesException(false)
.transform(df.coalesce(1))
.collect()
}
}

override def testObjects(): Seq[TestObject[TextAnalyze]] =
Seq(new TestObject[TextAnalyze](model, df))

override def reader: MLReadable[_] = TextAnalyze
}
//class TextAnalyzeSuite extends TransformerFuzzing[TextAnalyze] with TextEndpoint {
//
// import spark.implicits._
//
// implicit val doubleEquality: Equality[Double] = TolerantNumerics.tolerantDoubleEquality(1e-3)
//
// def df: DataFrame = Seq(
// ("en", "I had a wonderful trip to Seattle last week and visited Microsoft."),
// ("en", "Another document bites the dust"),
// (null, "ich bin ein berliner"),
// (null, null),
// ("en", null),
// ("invalid", "This is irrelevant as the language is invalid")
// ).toDF("language", "text")
//
// def model: TextAnalyze = new TextAnalyze()
// .setSubscriptionKey(textKey)
// .setLocation(textApiLocation)
// .setLanguageCol("language")
// .setOutputCol("response")
// .setErrorCol("error")
//
// def prepResults(df: DataFrame): Seq[Option[UnpackedTextAnalyzeResponse]] = {
// val fromRow = model.unpackedResponseBinding.makeFromRowConverter
// df.collect().map(row => Option(row.getAs[Row](model.getOutputCol)).map(fromRow)).toList
// }
//
// test("Basic usage") {
// val topResult = prepResults(model.transform(df.limit(1).coalesce(1))).head.get
// assert(topResult.pii.get.document.get.entities.head.text == "last week")
// assert(topResult.sentimentAnalysis.get.document.get.sentiment == "positive")
// assert(topResult.entityLinking.get.document.get.entities.head.dataSource == "Wikipedia")
// assert(topResult.keyPhraseExtraction.get.document.get.keyPhrases.head == "wonderful trip")
// assert(topResult.entityRecognition.get.document.get.entities.head.text == "trip")
// }
//
// test("Manual Batching") {
// val batchedDf = new FixedMiniBatchTransformer().setBatchSize(10).transform(df.coalesce(1))
// val resultDF = new FlattenBatch().transform(model.transform(batchedDf))
// val topResult = prepResults(resultDF).head.get
// assert(topResult.pii.get.document.get.entities.head.text == "last week")
// assert(topResult.sentimentAnalysis.get.document.get.sentiment == "positive")
// assert(topResult.entityLinking.get.document.get.entities.head.dataSource == "Wikipedia")
// assert(topResult.keyPhraseExtraction.get.document.get.keyPhrases.head == "wonderful trip")
// assert(topResult.entityRecognition.get.document.get.entities.head.text == "trip")
// }
//
// test("Large Batching") {
// val bigDF = (0 until 25).map(i => s"This is fantastic sentence number $i").toDF("text")
// val model2 = model.setLanguage("en").setBatchSize(25)
// val results = prepResults(model2.transform(bigDF.coalesce(1)))
// assert(results.length == 25)
// assert(results(24).get.sentimentAnalysis.get.document.get.sentiment == "positive")
// }
//
// test("Exceeded Retries Info") {
// val badModel = model
// .setPollingDelay(0)
// .setInitialPollingDelay(0)
// .setMaxPollingRetries(1)
//
// val results = badModel
// .setSuppressMaxRetriesException(true)
// .transform(df.coalesce(1))
// assert(results.where(!col("error").isNull).count() > 0)
//
// assertThrows[SparkException] {
// badModel.setSuppressMaxRetriesException(false)
// .transform(df.coalesce(1))
// .collect()
// }
// }
//
// override def testObjects(): Seq[TestObject[TextAnalyze]] =
// Seq(new TestObject[TextAnalyze](model, df))
//
// override def reader: MLReadable[_] = TextAnalyze
//}
2 changes: 1 addition & 1 deletion ...e/src/test/scala/com/microsoft/azure/synapse/ml/cognitive/translate/TranslatorSuite.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ class TranslateSuite extends TransformerFuzzing[Translate]

test("Translate with dynamic dictionary") {
val result1 = getTranslationTextResult(translate.setToLanguage(Seq("de")), textDf5).collect()
assert(result1(0).getSeq(0).mkString("\n") == "Das Wort wordomatic ist ein Wörterbucheintrag.")
assert(result1(0).getSeq(0).mkString("\n").contains("Das Wort"))
}

override def testObjects(): Seq[TestObject[Translate]] =
Expand Down
40 changes: 28 additions & 12 deletions core/src/main/scala/com/microsoft/azure/synapse/ml/logging/SynapseMLLogging.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,16 @@ package com.microsoft.azure.synapse.ml.logging

import com.microsoft.azure.synapse.ml.build.BuildInfo
import com.microsoft.azure.synapse.ml.logging.common.SASScrubber
import com.microsoft.azure.synapse.ml.logging.fabric.CertifiedEventClient.logToCertifiedEvents
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import spray.json.DefaultJsonProtocol._
import spray.json._

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

case class RequiredLogFields(uid: String,
className: String,
Expand Down Expand Up @@ -110,16 +113,29 @@ trait SynapseMLLogging extends Logging {

protected def logBase(methodName: String,
numCols: Option[Int],
executionSeconds: Option[Double]
executionSeconds: Option[Double],
logCertifiedEvent: Boolean = false
): Unit = {
logBase(getPayload(
methodName,
numCols,
executionSeconds,
None))
None), logCertifiedEvent)
}

protected def logBase(info: Map[String, String]): Unit = {
protected def logBase(info: Map[String, String], logCertifiedEvent: Boolean): Unit = {
if (logCertifiedEvent) {
Future {
logToCertifiedEvents(
info("libraryName"),
info("method"),
info -- Seq("libraryName", "method")
)
}.failed.map {
case e: Exception => logErrorBase("certifiedEventLogging", e)
}
}

logInfo(info.toJson.compactPrint)
}

Expand All @@ -130,23 +146,23 @@ trait SynapseMLLogging extends Logging {
}

def logClass(): Unit = {
logBase("constructor", None, None)
logBase("constructor", None, None, true)
}

def logFit[T](f: => T, columns: Int): T = {
logVerb("fit", f, Some(columns))
def logFit[T](f: => T, columns: Int, logCertifiedEvent: Boolean = true): T = {
logVerb("fit", f, columns, logCertifiedEvent)
}

def logTransform[T](f: => T, columns: Int): T = {
logVerb("transform", f, Some(columns))
def logTransform[T](f: => T, columns: Int, logCertifiedEvent: Boolean = true): T = {
logVerb("transform", f, columns, logCertifiedEvent)
}

def logVerb[T](verb: String, f: => T, columns: Option[Int] = None): T = {
def logVerb[T](verb: String, f: => T, columns: Int = -1, logCertifiedEvent: Boolean = false): T = {
val startTime = System.nanoTime()
try {
val ret = f
logBase(verb, columns, Some((System.nanoTime() - startTime) / 1e9))
ret
// Begin emitting certified event.
logBase(verb, Some(columns), Some((System.nanoTime() - startTime) / 1e9), logCertifiedEvent)
f
} catch {
case e: Exception =>
logErrorBase(verb, e)
Expand Down
Loading

0 comments on commit d0690a3

Please sign in to comment.