Static Scala -> Python converter, verified by scalac-vs-python differential runs
454
n8500x/scala2py converts Scala 2 code to runnable Python with a real parser (scalameta),
a Python ast emitter and a runtime shim (scala_compat), and proves the result:
verify compiles and runs the original with the bundled Scala compiler, runs the Python,
and diffs the outputs. The :spark tag adds PySpark so Spark programs — including code
generated by Informatica BDM/DEI's Spark engine — are diffed Scala-on-Spark vs PySpark.
docker run --rm -v "$PWD:/work" n8500x/scala2py convert src/main/scala -o out # Python + coverage report
docker run --rm -v "$PWD:/work" n8500x/scala2py verify src/main/scala # Scala vs Python diff
docker run --rm -v "$PWD:/work" n8500x/scala2py:spark verify mappings/ --timeout 300
docker run --rm n8500x/scala2py docs # everything baked in
Nothing is downloaded at run time (air-gap friendly). Anything the converter cannot
translate becomes an S.unsupported(...) call and a line in scala2py-report.md with
file, line and source — a work list, never a silent drop.
Informatica's Spark engine generates one object m_<MAPPING> per mapping: a SparkSession,
one DataFrame per transformation (source qualifiers, expressions, filters, joiners,
lookups, aggregators, routers, sorters, update strategies), ports named in_/out_/v_,
and calls into the Informatica function runtime exposed as UDFs. Below is the widest of
the four mock mappings bundled in the image — m_SALES_FACT_LOAD: 10 sources, a union
of two order channels, eight chained joins/lookups (inner and left with defaults), a
date-range promotion lookup, a 12-port expression stage, an error router, a five-key
aggregator, a pivot and a wide fact target. (The runtime jar is mocked as
InfaFunc/InfaUdf until a real generated file is available; every call to it is a
mapping-table entry.)
package infa.generated
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import infa.runtime.InfaUdf
// MOCK of Informatica BDM Spark-engine generated code for a WIDE mapping, m_SALES_FACT_LOAD:
// SQ_ONLINE_ORDERS ∪ SQ_STORE_ORDERS (UNION_ORDERS) -> JNR_LINES (inner, SQ_ORDER_LINES)
// -> LKP_PRODUCT -> LKP_CATEGORY -> LKP_CUSTOMER -> LKP_COUNTRY -> LKP_STORE (left)
// -> LKP_FX_RATE (left, default 1.0) -> LKP_PROMOTION (left, valid-date range)
// -> EXP_MEASURES -> RTR_QUALITY (VALID / ERROR) -> AGG_FACT (dept, category, region, channel, month)
// -> SRT_FACT -> TGT_SALES_FACT ; AGG_SEGMENT_PIVOT -> TGT_SEGMENT_CHANNEL ; TGT_ERR_ROWS
object m_SALES_FACT_LOAD {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.master("local[1]").appName("m_SALES_FACT_LOAD")
.config("spark.ui.enabled", "false").config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
// ---- sources (10 source qualifiers)
val SQ_ONLINE_ORDERS: DataFrame = spark.createDataFrame(Seq(
(1001, 1, "2024-01-05", "PROMO10", "AED"), (1002, 2, "2024-01-20", "", "USD"),
(1003, 3, "2024-02-02", "PROMO10", "AED"), (1004, 4, "2024-02-15", "", "GBP"),
(1005, 99, "2024-02-16", "", "AED")
)).toDF("ORDER_ID", "CUST_ID", "ORDER_DT", "PROMO_CD", "CURRENCY")
val SQ_STORE_ORDERS: DataFrame = spark.createDataFrame(Seq(
(2001, 1, 10, "2024-01-09", "AED"), (2002, 3, 20, "2024-03-01", "AED"),
(2003, 5, 30, "2024-03-05", "EUR"), (2004, 2, 10, "2024-03-30", "AED")
)).toDF("ORDER_ID", "CUST_ID", "STORE_ID", "ORDER_DT", "CURRENCY")
val SQ_ORDER_LINES: DataFrame = spark.createDataFrame(Seq(
(1001, 1, "P-1", 10, 2.5), (1001, 2, "P-2", 1, 21.0), (1002, 1, "P-3", 2, 149.0),
(1003, 1, "P-1", 100, 2.4), (1004, 1, "P-9", 1, 5.0), (1005, 1, "P-1", 3, 2.5),
(2001, 1, "P-2", 4, 21.0), (2002, 1, "P-3", 1, 149.0), (2002, 2, "P-5", 500, 0.05),
(2003, 1, "P-2", 2, 21.0), (2004, 1, "P-1", 50, 2.5), (2004, 2, "P-3", 1, 140.0)
)).toDF("ORDER_ID", "LINE_NO", "PRODUCT_CD", "QTY", "UNIT_PRICE")
val SQ_PRODUCTS: DataFrame = spark.createDataFrame(Seq(
("P-1", "Widget", "C-HW"), ("P-2", "Gadget", "C-HW"), ("P-3", "Gizmo", "C-EL"), ("P-5", "Bolt", "C-HW")
)).toDF("PRODUCT_CD", "PRODUCT_NAME", "CATEGORY_CD")
val SQ_CATEGORIES: DataFrame = spark.createDataFrame(Seq(
("C-HW", "Hardware", "Industrial"), ("C-EL", "Electronics", "Consumer")
)).toDF("CATEGORY_CD", "CATEGORY_NAME", "DEPT")
val SQ_CUSTOMERS: DataFrame = spark.createDataFrame(Seq(
(1, "ACME", "ENTERPRISE", "AE"), (2, "GLOBEX", "SMB", "US"), (3, "INITECH", "SMB", "AE"),
(4, "UMBRELLA", "ENTERPRISE", "GB"), (5, "HOOLI", "SMB", "DE")
)).toDF("CUST_ID", "CUST_NAME", "SEGMENT", "COUNTRY_CD")
val SQ_COUNTRIES: DataFrame = spark.createDataFrame(Seq(
("AE", "United Arab Emirates", "MEA"), ("US", "United States", "AMER"), ("GB", "United Kingdom", "EMEA"), ("DE", "Germany", "EMEA")
)).toDF("COUNTRY_CD", "COUNTRY_NAME", "REGION")
val SQ_STORES: DataFrame = spark.createDataFrame(Seq(
(10, "Dubai Mall", "MEA"), (20, "Marina", "MEA"), (30, "Berlin Mitte", "EMEA")
)).toDF("STORE_ID", "STORE_NAME", "STORE_REGION")
val SQ_FX_RATES: DataFrame = spark.createDataFrame(Seq(
("AED", 0.2723), ("GBP", 1.27), ("EUR", 1.08)
)).toDF("CURRENCY", "RATE_TO_USD")
val SQ_PROMOTIONS: DataFrame = spark.createDataFrame(Seq(
("PROMO10", 10.0, "2024-01-01", "2024-01-31"), ("PROMO25", 25.0, "2024-06-01", "2024-06-30")
)).toDF("PROMO_CD", "DISCOUNT_PCT", "VALID_FROM", "VALID_TO")
// ---- UNION_ORDERS: both channels into one shape
val online = SQ_ONLINE_ORDERS.select(col("ORDER_ID"), col("CUST_ID"), lit(null).cast("int").as("STORE_ID"),
col("ORDER_DT"), col("PROMO_CD"), col("CURRENCY"), lit("ONLINE").as("CHANNEL"))
val store = SQ_STORE_ORDERS.select(col("ORDER_ID"), col("CUST_ID"), col("STORE_ID"),
col("ORDER_DT"), lit("").as("PROMO_CD"), col("CURRENCY"), lit("STORE").as("CHANNEL"))
val UNION_ORDERS = online.union(store)
// ---- joins and lookups (one DataFrame per transformation, as generated)
val JNR_LINES = UNION_ORDERS.join(SQ_ORDER_LINES, Seq("ORDER_ID"), "inner")
val LKP_PRODUCT = JNR_LINES.join(SQ_PRODUCTS, Seq("PRODUCT_CD"), "left")
val LKP_CATEGORY = LKP_PRODUCT.join(SQ_CATEGORIES, Seq("CATEGORY_CD"), "left")
val LKP_CUSTOMER = LKP_CATEGORY.join(SQ_CUSTOMERS, Seq("CUST_ID"), "left")
val LKP_COUNTRY = LKP_CUSTOMER.join(SQ_COUNTRIES, Seq("COUNTRY_CD"), "left")
val LKP_STORE = LKP_COUNTRY.join(SQ_STORES, Seq("STORE_ID"), "left")
val LKP_FX_RATE = LKP_STORE.join(SQ_FX_RATES, Seq("CURRENCY"), "left")
.withColumn("RATE_TO_USD", coalesce(col("RATE_TO_USD"), lit(1.0)))
val LKP_PROMOTION = LKP_FX_RATE.join(SQ_PROMOTIONS, Seq("PROMO_CD"), "left")
.withColumn("PROMO_VALID", col("DISCOUNT_PCT").isNotNull && col("ORDER_DT").between(col("VALID_FROM"), col("VALID_TO")))
// ---- EXP_MEASURES
val EXP_MEASURES = LKP_PROMOTION
.withColumn("v_GROSS", col("QTY") * col("UNIT_PRICE"))
.withColumn("v_DISC_PCT", when(col("PROMO_VALID"), col("DISCOUNT_PCT")).otherwise(lit(0.0)))
.withColumn("v_NET", col("v_GROSS") * (lit(1.0) - col("v_DISC_PCT") / 100.0))
.withColumn("out_GROSS_USD", InfaUdf.ROUND2(col("v_GROSS") * col("RATE_TO_USD")))
.withColumn("out_NET_USD", InfaUdf.ROUND2(col("v_NET") * col("RATE_TO_USD")))
.withColumn("out_MONTH", substring(col("ORDER_DT"), 1, 7))
.withColumn("out_REGION", coalesce(col("STORE_REGION"), col("REGION"), lit("UNKNOWN")))
.withColumn("out_STORE_NAME", InfaUdf.NVL(col("STORE_NAME"), lit("ONLINE")))
.withColumn("out_ERR", when(col("PRODUCT_NAME").isNull, lit("UNKNOWN PRODUCT"))
.when(col("CUST_NAME").isNull, lit("UNKNOWN CUSTOMER")).otherwise(lit(null).cast("string")))
// ---- RTR_QUALITY
val RTR_VALID = EXP_MEASURES.filter(col("out_ERR").isNull)
val RTR_ERROR = EXP_MEASURES.filter(col("out_ERR").isNotNull)
// ---- AGG_FACT + SRT_FACT -> TGT_SALES_FACT
val AGG_FACT = RTR_VALID.groupBy("DEPT", "CATEGORY_NAME", "out_REGION", "CHANNEL", "out_MONTH").agg(
count("LINE_NO").as("LINE_CNT"),
sum("QTY").as("QTY"),
round(sum("out_GROSS_USD"), 2).as("GROSS_USD"),
round(sum("out_NET_USD"), 2).as("NET_USD"),
countDistinct("CUST_ID").as("CUSTOMERS"))
val SRT_FACT = AGG_FACT.orderBy("DEPT", "CATEGORY_NAME", "out_REGION", "CHANNEL", "out_MONTH")
println("TGT_SALES_FACT:")
SRT_FACT.show(100, false)
// ---- AGG_SEGMENT_PIVOT -> TGT_SEGMENT_CHANNEL
val AGG_SEGMENT_PIVOT = RTR_VALID.groupBy("SEGMENT").pivot("CHANNEL", Seq("ONLINE", "STORE"))
.agg(round(sum("out_NET_USD"), 2)).na.fill(0.0).orderBy("SEGMENT")
println("TGT_SEGMENT_CHANNEL:")
AGG_SEGMENT_PIVOT.show()
// ---- TGT_ERR_ROWS
println("TGT_ERR_ROWS:")
RTR_ERROR.select("ORDER_ID", "LINE_NO", "PRODUCT_CD", "CUST_ID", "out_ERR").orderBy("ORDER_ID", "LINE_NO")
.collect().foreach(r => println(" " + r.mkString("|")))
val valid = RTR_VALID.count()
val errors = RTR_ERROR.count()
println(s"lines in=${SQ_ORDER_LINES.count()} valid=$valid errors=$errors fact rows=${AGG_FACT.count()}")
spark.stop()
}
}
scala2py convert, unedited)# Generated by scala2py 0.1.0 from m_SALES_FACT_LOAD.scala
# Runtime support: scala_compat (imported as S).
from __future__ import annotations
import sys
import scala_compat as S
from pyspark.sql import DataFrame
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
class _m_SALES_FACT_LOAD(S.ScalaObject):
def main(self, args: list) -> None:
spark = (
SparkSession.builder.master("local[1]")
.appName("m_SALES_FACT_LOAD")
.config("spark.ui.enabled", "false")
.config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
SQ_ONLINE_ORDERS = spark.createDataFrame(
S.Seq(
(1001, 1, "2024-01-05", "PROMO10", "AED"),
(1002, 2, "2024-01-20", "", "USD"),
(1003, 3, "2024-02-02", "PROMO10", "AED"),
(1004, 4, "2024-02-15", "", "GBP"),
(1005, 99, "2024-02-16", "", "AED"),
),
T.StructType(
[
T.StructField("_1", T.IntegerType(), False),
T.StructField("_2", T.IntegerType(), False),
T.StructField("_3", T.StringType(), True),
T.StructField("_4", T.StringType(), True),
T.StructField("_5", T.StringType(), True),
]
),
).toDF("ORDER_ID", "CUST_ID", "ORDER_DT", "PROMO_CD", "CURRENCY")
# ... SQ_STORE_ORDERS, SQ_ORDER_LINES, SQ_PRODUCTS, SQ_CATEGORIES, SQ_CUSTOMERS, SQ_COUNTRIES,
# ... SQ_STORES, SQ_FX_RATES, SQ_PROMOTIONS: nine more sources, same shape (elided here only;
# ... the full exported file is in the image: examples/informatica-spark/converted/)
online = SQ_ONLINE_ORDERS.select(
F.col("ORDER_ID"),
F.col("CUST_ID"),
F.lit(None).cast("int").alias("STORE_ID"),
F.col("ORDER_DT"),
F.col("PROMO_CD"),
F.col("CURRENCY"),
F.lit("ONLINE").alias("CHANNEL"),
)
store = SQ_STORE_ORDERS.select(
F.col("ORDER_ID"),
F.col("CUST_ID"),
F.col("STORE_ID"),
F.col("ORDER_DT"),
F.lit("").alias("PROMO_CD"),
F.col("CURRENCY"),
F.lit("STORE").alias("CHANNEL"),
)
UNION_ORDERS = online.union(store)
JNR_LINES = UNION_ORDERS.join(SQ_ORDER_LINES, S.Seq("ORDER_ID"), "inner")
LKP_PRODUCT = JNR_LINES.join(SQ_PRODUCTS, S.Seq("PRODUCT_CD"), "left")
LKP_CATEGORY = LKP_PRODUCT.join(SQ_CATEGORIES, S.Seq("CATEGORY_CD"), "left")
LKP_CUSTOMER = LKP_CATEGORY.join(SQ_CUSTOMERS, S.Seq("CUST_ID"), "left")
LKP_COUNTRY = LKP_CUSTOMER.join(SQ_COUNTRIES, S.Seq("COUNTRY_CD"), "left")
LKP_STORE = LKP_COUNTRY.join(SQ_STORES, S.Seq("STORE_ID"), "left")
LKP_FX_RATE = LKP_STORE.join(SQ_FX_RATES, S.Seq("CURRENCY"), "left").withColumn(
"RATE_TO_USD", F.coalesce(F.col("RATE_TO_USD"), F.lit(1.0))
)
LKP_PROMOTION = LKP_FX_RATE.join(SQ_PROMOTIONS, S.Seq("PROMO_CD"), "left").withColumn(
"PROMO_VALID",
F.col("DISCOUNT_PCT").isNotNull()
& F.col("ORDER_DT").between(F.col("VALID_FROM"), F.col("VALID_TO")),
)
EXP_MEASURES = (
LKP_PROMOTION.withColumn("v_GROSS", F.col("QTY") * F.col("UNIT_PRICE"))
.withColumn(
"v_DISC_PCT",
F.when(F.col("PROMO_VALID"), F.col("DISCOUNT_PCT")).otherwise(F.lit(0.0)),
)
.withColumn(
"v_NET", F.col("v_GROSS") * (F.lit(1.0) - S.div(F.col("v_DISC_PCT"), 100.0))
)
.withColumn("out_GROSS_USD", InfaUdf.ROUND2(F.col("v_GROSS") * F.col("RATE_TO_USD")))
.withColumn("out_NET_USD", InfaUdf.ROUND2(F.col("v_NET") * F.col("RATE_TO_USD")))
.withColumn("out_MONTH", F.substring(F.col("ORDER_DT"), 1, 7))
.withColumn(
"out_REGION", F.coalesce(F.col("STORE_REGION"), F.col("REGION"), F.lit("UNKNOWN"))
)
.withColumn("out_STORE_NAME", InfaUdf.NVL(F.col("STORE_NAME"), F.lit("ONLINE")))
.withColumn(
"out_ERR",
F.when(F.col("PRODUCT_NAME").isNull(), F.lit("UNKNOWN PRODUCT"))
.when(F.col("CUST_NAME").isNull(), F.lit("UNKNOWN CUSTOMER"))
.otherwise(F.lit(None).cast("string")),
)
)
RTR_VALID = S.seq(EXP_MEASURES).filter(F.col("out_ERR").isNull())
RTR_ERROR = S.seq(EXP_MEASURES).filter(F.col("out_ERR").isNotNull())
AGG_FACT = (
S.seq(RTR_VALID)
.groupBy("DEPT", "CATEGORY_NAME", "out_REGION", "CHANNEL", "out_MONTH")
.agg(
F.count("LINE_NO").alias("LINE_CNT"),
F.sum("QTY").alias("QTY"),
F.round(F.sum("out_GROSS_USD"), 2).alias("GROSS_USD"),
F.round(F.sum("out_NET_USD"), 2).alias("NET_USD"),
F.countDistinct("CUST_ID").alias("CUSTOMERS"),
)
)
SRT_FACT = AGG_FACT.orderBy("DEPT", "CATEGORY_NAME", "out_REGION", "CHANNEL", "out_MONTH")
S.println("TGT_SALES_FACT:")
SRT_FACT.show(100, truncate=False)
AGG_SEGMENT_PIVOT = (
S.seq(RTR_VALID)
.groupBy("SEGMENT")
.pivot("CHANNEL", S.Seq("ONLINE", "STORE"))
.agg(F.round(F.sum("out_NET_USD"), 2))
.na.fill(0.0)
.orderBy("SEGMENT")
)
S.println("TGT_SEGMENT_CHANNEL:")
AGG_SEGMENT_PIVOT.show()
S.println("TGT_ERR_ROWS:")
S.seq(
S.seq(
RTR_ERROR.select("ORDER_ID", "LINE_NO", "PRODUCT_CD", "CUST_ID", "out_ERR").orderBy(
"ORDER_ID", "LINE_NO"
)
).collect()
).foreach(lambda r: S.println(f" {S.mkString(r, '|')}"))
valid = S.seq(RTR_VALID).count()
errors = S.seq(RTR_ERROR).count()
S.println(
f"lines in={S.seq(SQ_ORDER_LINES).count()} valid={valid} errors={errors} fact rows={S.seq(AGG_FACT).count()}"
)
spark.stop()
m_SALES_FACT_LOAD = S.lazy_object(_m_SALES_FACT_LOAD)
from infa.runtime.InfaRuntimeMock import InfaUdf
if __name__ == "__main__":
sys.setrecursionlimit(20000)
m_SALES_FACT_LOAD.main(S.args_list(sys.argv[1:]))
verify: MATCH)TGT_SALES_FACT:
+----------+-------------+----------+-------+---------+--------+---+---------+-------+---------+
|DEPT |CATEGORY_NAME|out_REGION|CHANNEL|out_MONTH|LINE_CNT|QTY|GROSS_USD|NET_USD|CUSTOMERS|
+----------+-------------+----------+-------+---------+--------+---+---------+-------+---------+
|Consumer |Electronics |AMER |ONLINE |2024-01 |1 |2 |298.0 |298.0 |1 |
|Consumer |Electronics |MEA |STORE |2024-03 |2 |2 |78.69 |78.69 |2 |
|Industrial|Hardware |EMEA |STORE |2024-03 |1 |2 |45.36 |45.36 |1 |
|Industrial|Hardware |MEA |ONLINE |2024-01 |2 |11 |12.53 |11.28 |1 |
|Industrial|Hardware |MEA |ONLINE |2024-02 |1 |100|65.35 |65.35 |1 |
|Industrial|Hardware |MEA |STORE |2024-01 |1 |4 |22.87 |22.87 |1 |
|Industrial|Hardware |MEA |STORE |2024-03 |2 |550|40.85 |40.85 |2 |
+----------+-------------+----------+-------+---------+--------+---+---------+-------+---------+
TGT_SEGMENT_CHANNEL:
+----------+------+-----+
| SEGMENT|ONLINE|STORE|
+----------+------+-----+
|ENTERPRISE| 11.28|22.87|
| SMB|363.35|164.9|
+----------+------+-----+
TGT_ERR_ROWS:
1004|1|P-9|4|UNKNOWN PRODUCT
1005|1|P-1|99|UNKNOWN CUSTOMER
lines in=12 valid=10 errors=2 fact rows=7
What the converter did: functions._/types._ imports → F/T; ===, =!=, &&,
||, ! on Columns → ==, !=, &, |, ~; .as("x") → .alias; udf(f _) →
F.udf(f, <declared return type>); show(100, false) → show(100, truncate=False);
createDataFrame(Seq((1, "a"), …)) gets an explicit StructType so column types and
nullability match Scala's inference; Seq("k") join keys, Window, pivot, na.fill,
coalesce, between, cast pass through; S.seq(x) is an identity for anything that is
not a String (it gives Strings the Seq[Char] API), S.div keeps JVM division semantics.
Known Spark-specific limits: Python UDF result columns are always nullable and run
slower than JVM UDFs — map known runtime functions to native F.* via --mappings;
the session must be named spark for Seq(...).toDF(...); the typed Dataset[T]
API (case-class encoders) is not covered.
Example (scala2py examples NAME --to DIR copies, converts and runs it) | Shows |
|---|---|
jdbc-report | DriverManager, PreparedStatement with ? params, batches, transactions, a JOIN … GROUP BY through a ResultSet loop, rows into case classes joined in Scala (SQLite bundled on both sides) |
collections-join | lookup Maps, for-comprehension inner/left joins over case classes, groupBy aggregates, formatted report |
csv-pipeline | PrintWriter/Source.fromFile CSV out and in, parsing, join, report file |
informatica-spark | the four mock mappings (m_CUSTOMER_LOAD, m_ORDER_SUMMARY, m_SCD2_DIM_LOAD, m_SALES_FACT_LOAD) — :spark image |
The exported Python of every example is baked at examples/<name>/converted/;
scala2py docs --to DIR copies the full README, the examples, the test-suite and the
reports out of the image.
Verification: selftest runs 44 differential tests — 9 syntax programs, a script-mode
fixture, two GitHub projects verbatim (a Lisp interpreter, TheAlgorithms/Scala), the four
examples above — each byte-identical to scalac. A breadth run over 1,287 Rosetta Code
Scala programs converts 95 % of files completely and makes 62 % of the runnable sample
byte-identical; the remaining gaps (implicit conversions, type-level code, unmapped Java
libraries) are listed in the report baked into the image (scala2py docs ROSETTA).
Full documentation: scala2py docs README.
Content type
Image
Digest
sha256:9e221cb22…
Size
158.2 MB
Last updated
25 days ago
docker pull n8500x/scala2py