@Hubert Dudek​ It works like that. I have one more question. How can I include and delete that query?

  microBatchOutputDF._jdf.sparkSession().sql("""
    MERGE INTO old o
    USING updates u
    ON u.id= o.id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

Or how I can add and delete rows from this pipeline.