afranzi

@udf(returnType=ArrayType(StringType()))
def to_upper_list(s):
    return [i.upper() for i in s]

# Case 1 - UDF annotation
to_upper_list(['potato', 'carrot', 'tomato'])

"""
TypeError: Invalid argument, not a string or column: ['potato', 'carrot', 'tomato'] of type <class 'list'>. 
For column literals, use 'lit', 'array', 'struct' or 'create_map' function

afranzi

from unittest import TestCase

import pytest
from pyspark.sql.types import StringType


@pytest.fixture(scope='function', autouse=True)
def mock_udf_annotation(monkeypatch):
    def dummy_udf(f):
        return f

afranzi

from unittest import TestCase

def dummy_udf(f):
    return f

def mock_udf(f=None, returnType=None):
    return f if f else dummy_udf

from mock import patch
patch('pyspark.sql.functions.udf', mock_udf).start()

afranzi

from importlib import reload
from unittest import TestCase

from mock import patch
from our_package.spark import udfs as UDF


def dummy_udf(f):
    return f

afranzi

import SchemaValidationListener._
import org.everit.json.schema.Schema
import org.everit.json.schema.event.{CombinedSchemaMatchEvent, SchemaReferencedEvent, ValidationListener}

import scala.collection.mutable.ListBuffer
import scala.util.matching.Regex
import scala.collection.JavaConverters._

object SchemaValidationListener {

afranzi

def getSchemaRef(event: JSONObject): String = {
  JsonObjectFields.getSchemaRef(event, defaultSchema)
} 
 
def validateEvent(schema: Schema, event: JSONObject): ValidationResult[JSONObject] = {
  val validationListener: SchemaValidationListener = SchemaValidationListener()
  val validator: Validator = Validator
    .builder
    .withListener(validationListener)
    .build()

afranzi

val schemaCache: LoadingCache[String, Schema] = CacheBuilder
  .newBuilder
  .maximumSize(MaximumCacheSize)
  .expireAfterAccess(CacheMinutes, TimeUnit.MINUTES)
  .build(new CacheLoader[String, Schema]() {
    override def load(schemaRef: String): Schema = {
      logger.info(s"Loading schema $schemaRef")
      loadSchema(schemaRef: String)
    }
  })

afranzi

import SchemaValidationListener._
import org.everit.json.schema.Schema
import org.everit.json.schema.event.{CombinedSchemaMatchEvent, SchemaReferencedEvent, ValidationListener}
 
import scala.collection.mutable.ListBuffer
import scala.util.matching.Regex
import scala.collection.JavaConverters._
 
object SchemaValidationListener {
 

afranzi

{
  "user": {
    "id": .userId
  },
  "device": {
    "id": "undefined",
    "platform": if(.p == "a") "Android" else "iPhone"
  },
  "product": {
    "id": "remix",

afranzi

def evolve(event: ValidationResult[JsonNode]): EvolutionResult[JsonNode] = {
  val schemasReferenced: Seq[SchemaReferenced] = event.schemasReferenced
  val json = event.event
 
  val schemasToEvolve = schemasReferenced
    .filter { case SchemaReferenced(_, schemaRef) => hasEvolution(schemaRef) }
 
  val eventEvolved = schemasToEvolve
    .foldLeft(json) {
      case (jsonEvent: JsonNode, SchemaReferenced(location, schemaRef)) =>