如何使用 Azure Synapse 在 Databricks 上删除表或删除行？

Question

我想使用 DROP TABLE 删除我的表，但我的 Databricks 上出现错误 JDBC，但如果我使用 SELECT，它不会出现任何错误，我有 2 个不同的函数可以 read_from_synapse和write_to_synapse。

这是我的read_from_synapse，我认为DROP TABLE可以使用查询

def read_from_synapse(sql, url, tempDir):
       df = (spark.read
              .format("com.databricks.spark.sqldw")
              .option("url", url)
              .option("tempDir", tempDir)
              .option("forwardSparkAzureStorageCredentials", "true")
              .option("query", sql)
              .load()
       )
       
       return df

这是我的write_to_synapse 可以插入表格的脚本

def write_to_synapse(df, table, write_mode, url, tempDir):
       df.write
              .format("com.databricks.spark.sqldw")
              .option("tableOptions", "CLUSTERED COLUMNSTORE INDEX, DISTRIBUTION = ROUND_ROBIN") # Added at 20200121
              .option("url", url)
              .option("dbtable", table)
              .option("forward_spark_azure_storage_credentials","True")
              .option("tempdir", tempDir)         
              .mode(write_mode)
              .save()

更新

我已经尝试过这个链接，但它并没有解决我的问题，它得到了这样的错误

com.databricks.spark.sqldw.SqlDWSideException: Azure Synapse Analytics failed to execute the JDBC query produced by the connector.

这从错误中更详细：

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<command-1485523310718777> in <module>
----> 1 drop_sdf_sales = azSynapse._delete_from_synapse(drop_sql, tempDir=temp_read_dir)

<command-1485523310718391> in _delete_from_synapse(self, sql, url, tempDir)
      5         if not tempDir:
      6             tempDir = self.azblob._get_blob_path('04-publish', 'sqlDwReadTempDirs')
----> 7         df = UtilAzSynapse.read_from_synapse(sql, url, tempDir)
      8         return df

<command-2362013028695578> in read_from_synapse(sql, url, tempDir)
     16                   .option("tempDir", tempDir)
     17                   .option("forwardSparkAzureStorageCredentials", "true")
---> 18                   .option("query", sql)
     19                   .load()
     20              )

/databricks/spark/python/pyspark/sql/readwriter.py in load(self, path, format, schema, **options)
    182             return self._df(self._jreader.load(self._spark._sc._jvm.PythonUtils.toSeq(path)))
    183         else:

Answer 1

使用 Azure Databricks 从 Azure Synapse 表中删除一行：

postActionsSQL = "DELETE from Persons WHERE PersonID=3"

df.write.format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://synapse.sql.azuresynapse.net:1433;database=master;user=master@synapse;password=XXXXXXX;encrypt=true;trustServerCertificate=true;hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;") \
  .option("tempDir", "wasbs://sampledata@synapse.blob.core.windows.net/temp") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "Persons") \
  .option("postActions",postActionsSQL) \
  .mode("overwrite") \
  .save()

使用 Azure Databricks 从 Azure Synapse 表中删除表：

postActionsSQL = "DROP TABLE Trip003"

df.write.format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://synapse.sql.azuresynapse.net:1433;database=master;user=master@synapse;password=XXXXXXX;encrypt=true;trustServerCertificate=true;hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;") \
  .option("tempDir", "wasbs://sampledata@synapse.blob.core.windows.net/temp") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "Trip003") \
  .option("postActions",postActionsSQL) \
  .mode("overwrite") \
  .save()