PySpark - Method andTopicNameStrategy([class java.lang.String, class java.lang.Boolean]) does not exist

[sai3563]

Hello,

I recently found out about ABRiS and I am trying to register schema to confluent schema registry after interpreting it from spark dataframe.

def to_avro_abris_config(config_map, topic, avro_schema):
    jvm_gateway = SparkContext._active_spark_context._gateway.jvm
    scala_map = jvm_gateway.PythonUtils.toScalaMap(config_map)

    # Provide and register the schema
    return jvm_gateway.za.co.absa.abris.config \
        .AbrisConfig \
        .toConfluentAvro() \
        .provideAndRegisterSchema(avro_schema) \
        .andTopicNameStrategy(topic, True) \
        .usingSchemaRegistry(scala_map)

jvm_gateway = SparkContext._active_spark_context._gateway.jvm
avro_schema_utils = jvm_gateway.za.co.absa.abris.avro.parsing.utils.AvroSchemaUtils
avro_schema = avro_schema_utils.toAvroSchema(df._jdf)

print(avro_schema.toString())

# Create ABRiS configuration
to_avro_abris_settings = to_avro_abris_config(config_map, topic, avro_schema.toString())

Even though the documentation says that andTopicNameStrategy is legit, I get

    raise Py4JError(
py4j.protocol.Py4JError: An error occurred while calling o325.andTopicNameStrategy. Trace:
py4j.Py4JException: Method andTopicNameStrategy([class java.lang.String, class java.lang.Boolean]) does not exist
	at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:321)
	at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:329)
	at py4j.Gateway.invoke(Gateway.java:274)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
	at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
	at java.base/java.lang.Thread.run(Thread.java:829)

What could be the issue here?

The record name strategy works, but I want to use topic name strategy.

PySpark 3.4.1
za.co.absa:abris_2.12:6.3.0

[cerveada]

In this case, it's usingTopicNameStrategy. You register schema using the strategy, that is the logic behind it, but understand it may be a bit confusing.

[sai3563]

@cerveada Thanks. Is there example code in python to upload pyspark dataframe to confluent topic which also involves the magic byte?

In the documentation, I could only find example to read from kafka. I'm looking for code in batch mode, not stream mode.

[cerveada]

None I am aware of, but it should be simple to create.

To tell Abris to create a magic byte, you just need to use .toConfluentAvro() in config builder.

Streaming or not should not make a difference. Abris just modifies the data frame column and from there it's up to you to send it to kafka or elsewhere.

[sai3563]

@cerveada I tried to write dataframe using Abris for 2 hours but failed. I'm not a java/scala expert, I come from R & Python. I just kept running into one error after the other when it comes to serialization & gave up. A simple example would be really helpful.

If not, then I completely understand. I'm about to start learning Java from tomorrow just for this as I tried a multitude of ways for 4 days straight (including this), but failed using python (pyspark). I'll have to deliver this by next week one way or the other, so ya, learn & adapt to survive.