ottomata / example_usage.py

Last active September 24, 2024 05:13

fsspec GitLabFileSystem

	from fsspec_gitlab import *
	import fsspec

	# Instantiate a gitlab repo filesystem
	gitlab_repo = fsspec.filesystem("gitlab", base_uri="https://gitlab.wikimedia.org", project_path="repos/data-engineering/workflow_utils")

	# list a directory
	gitlab_repo.ls("workflow_utils", detail=True)

	# open and read a file

ottomata / mediawiki_cdc_to_paimon_demo.md

Last active October 16, 2024 17:26

Apache Paimon + MediaWIki demo

Apache Paimon + MediaWiki demo

Setup

ottomata / _setup.sh

Last active January 16, 2024 14:12

	cd ~/
	mkdir flink-sql-libs
	cd flink-sql-libs/

	wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/1.17.2/flink-connector-kafka-1.17.2.jar
	wget https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.4.0/kafka-clients-3.4.0.jar
	# Only need this if querying WMF event streams.
	# https://wikitech.wikimedia.org/wiki/Event_Platform/Stream_Processing/Flink_Catalog#Creating_Tables
	# wget https://archiva.wikimedia.org/repository/releases/org/wikimedia/eventutilities-flink/1.3.3/eventutilities-flink-1.3.3-jar-with-dependencies.jar

ottomata / output.txt

Created January 2, 2024 21:55

EvolveHiveTable with comments output

	spark3-submit --class org.wikimedia.analytics.refinery.job.refine.tool.EvolveHiveTable ./refinery-job/target/refinery-job-0.2.28-SNAPSHOT-shaded.jar --table=event.mediawiki_page_change_v1 --schema_uri=/mediawiki/page/change/latest --dry_run=true

	24/01/02 21:49:53 INFO DataFrameToHive: Found difference in schemas for Hive table otto.mw_page_change0
	Table schema:
	root
	-- _schema: string (nullable = true)
	-- changelog_kind: string (nullable = true)
	-- comment: string (nullable = true)
	-- created_redirect_page: struct (nullable = true)
	\|-- is_redirect: boolean (nullable = true)

ottomata / pyflink_output_tag_reuse_fail_word_count_example.py

Last active February 14, 2023 19:50

Example showing failure when re-using an OutputTag in multiple ProcessFunctions


	import argparse
	import logging
	import sys

	from pyflink.common import WatermarkStrategy, Encoder, Types
	from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, ProcessFunction, OutputTag
	from pyflink.datastream.connectors.file_system import FileSource, StreamFormat, FileSink, OutputFileConfig, RollingPolicy

ottomata / pyflink_sideout_fail_word_count_example.py

Last active February 10, 2023 14:05

Example showing failure when not calling get_side_output on a pyflink DataStream immediately after the ProcessFunction that emits to a side output


	import argparse
	import logging
	import sys

	from pyflink.common import WatermarkStrategy, Encoder, Types
	from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, ProcessFunction, OutputTag
	from pyflink.datastream.connectors.file_system import FileSource, StreamFormat, FileSink, OutputFileConfig, RollingPolicy

ottomata / Dockerfile

Last active December 8, 2022 18:48

flink-kubernetes-operator python example

	FROM docker-registry.wikimedia.org/flink:1.16.0-37


	# add python script
	USER root
	RUN mkdir -p /srv/flink_app && ls
	ADD python_demo.py /srv/flink_app/python_demo.py
	USER flink

ottomata / schema_classes.py

Created November 17, 2022 18:23

datahub schema classes

	# flake8: noqa

	# This file is autogenerated by /metadata-ingestion/scripts/avro_codegen.py
	# Do not modify manually!

	# pylint: skip-file
	# fmt: off

	# The SchemaFromJSONData method only exists in avro-python3, but is called make_avsc_object in avro.
	# We can use this fact to detect conflicts between the two packages. Pip won't detect those conflicts

ottomata / 0_flink_sql_enrich_demo.sh

Created November 7, 2022 20:38

Mediawiki Content Flink Streaming SQL Content Enrich

	# Need to download flink-connector-kafka-1.15.2.jar and kafka-clients-2.4.1.jar
	./bin/sql-client.sh -i flink_sql_init.sql -pyfs get_revision_content_udf.py -pyexec /home/otto/pyflink_udf2/bin/python3 -pyclientexec /home/otto/pyflink_udf2/bin/python3 -j /home/otto/flink-connector-kafka-1.15.2.jar -j /home/otto/kafka-clients-2.4.1.jar

ottomata / enrich.sql

Created November 3, 2022 21:50

flink SQL enrichment UDF experiements 2



	CREATE TEMPORARY TABLE mediawiki_page_change (
	`wiki_id` STRING,
	`meta` ROW<domain STRING>,
	`page_change_kind` STRING,
	`page` ROW<page_id BIGINT, page_title STRING>,
	`revision` ROW<rev_id BIGINT, content_slots MAP<string, ROW<slot_role STRING, content_format STRING, content_body STRING>>>
	) WITH (
	'connector' = 'kafka',