Mettre en œuvre des fonctions définies par l'utilisateur

Le service géré pour Apache Flink Studio utilise la terminologie d’Apache Zeppelin selon laquelle un bloc-notes est une instance Zeppelin pouvant contenir plusieurs notes. Chaque note peut ensuite contenir plusieurs paragraphes. Avec le service géré pour Apache Flink Studio, le processus d’interprétation est partagé entre toutes les notes du bloc-notes. Ainsi, si vous effectuez un enregistrement de fonction explicite à l'aide de createTemporarySystemFunction dans une note, celle-ci peut être référencée telle quelle dans une autre note du même bloc-notes.

L’opération Déployer en tant qu’application fonctionne toutefois sur une note individuelle et non sur toutes les notes du bloc-notes. Lorsque vous effectuez un déploiement en tant qu’application, seul le contenu d’une note active est utilisé pour générer l’application. Tout enregistrement de fonction explicite effectué dans d’autres blocs-notes ne fait pas partie des dépendances d’application générées. En outre, lors de l'option Déployer en tant qu'application, un enregistrement de fonction implicite se produit en convertissant le nom de classe principal JAR de en une chaîne minuscule.

Par exemple, s'il s'TextAnalyticsUDFagit de la classe principale pour UDFJAR, un enregistrement implicite entraînera le nom de la fonctiontextanalyticsudf. Ainsi, si un enregistrement de fonction explicite dans la note 1 de Studio se produit comme suit, toutes les autres notes de ce bloc-notes (disons la note 2) peuvent faire référence à la fonction par son nom myNewFuncNameForClass grâce à l’interprète partagé :

stenv.createTemporarySystemFunction("myNewFuncNameForClass", new TextAnalyticsUDF())

Toutefois, lors de l’opération de déploiement en tant qu’application mentionnée à la note 2, cet enregistrement explicite ne sera pas inclus dans les dépendances et, par conséquent, l’application déployée ne fonctionnera pas comme prévu. En raison de l’enregistrement implicite, par défaut, toutes les références à cette fonction sont supposées être avec textanalyticsudf et non myNewFuncNameForClass.

S’il est nécessaire d’enregistrer un nom de fonction personnalisé, la note 2 elle-même devrait contenir un autre paragraphe pour effectuer un autre enregistrement explicite comme suit :

%flink(parallelism=l)
import com.amazonaws.kinesis.udf.textanalytics.TextAnalyticsUDF 
# re-register the JAR for UDF with custom name
stenv.createTemporarySystemFunction("myNewFuncNameForClass", new TextAnalyticsUDF())

%flink. ssql(type=update, parallelism=1) 
INSERT INTO
    table2
SELECT
    myNewFuncNameForClass(column_name)
FROM
    table1
;

Si vous UDF JAR incluez FlinkSDKs, configurez votre projet Java de manière à ce que le code UDF source puisse être compilé par rapport au FlinkSDKs, mais que les SDK classes Flink ne soient pas elles-mêmes incluses dans l'artefact de construction, par exemple le. JAR

Vous pouvez utiliser provided scope dans Apache Maven, les déclarations de compileOnly dépendance dans Gradle, provided scope in SBT ou une directive équivalente dans la configuration de construction de leur UDF projet. Vous pouvez vous référer à ce pom à partir de l'exemple du fichier UDF jar, qui répond à une telle condition préalable sur un projet maven. Pour un step-by-step didacticiel complet, consultez ce guide Translate, rédigez et analysez les données de streaming à l'aide des SQL fonctions d'Amazon Managed Service pour Apache Flink, Amazon Translate et Amazon Comprehend.