ADF 映射数据流 - Sink 转换动态分区数

Question

我有以下表达式来计算 Sink 转换中的 "Number of partitions" 作为动态内容，

toInteger (round( iif(toDecimal('5671478512', 38, 2) <= 104857600, toDecimal(1.00) , toDecimal('5671478512', 38, 2)/104857600) ) )

此表达式的结果必须是整数 54，但出于某种原因，它在 ADF 门户中调试时会抛出错误。

当我在派生列转换中尝试精确表达式时，我得到预期值 54。

知道为什么它在 "Number of partitions" 中失败了吗？但在派生列中测试时有效

以下是我在 "Number of partitions" 动态内容

中添加表达式时出现的错误

collectPreviewData failure on job=e97f7e77-abae-41f2-95dd-7d2d0e03800b, jobState=Failed com.microsoft.dataflow.Issues: DF-SYS-01 - requirement failed: Number of partitions (0) must be positive. - Nonejava.lang.IllegalArgumentException: requirement failed: Number of partitions (0) must be positive.
    at scala.Predef$.require(Predef.scala:224)
    at org.apache.spark.sql.catalyst.plans.logical.RepartitionByExpression.<init>(basicLogicalOperators.scala:1123)
    at com.microsoft.dataflow.TransformPlanner$$anonfun$physicalPartitionPlan.apply(Transformer.scala:299)
    at com.microsoft.dataflow.TransformPlanner$$anonfun$physicalPartitionPlan.apply(Transformer.scala:283)
    at scala.collection.immutable.Stream.map(Stream.scala:418)
    at com.microsoft.dataflow.TransformPlanner$class.physicalPartitionPlan(Transformer.scala:283)
    at com.microsoft.dataflow.transformers.ExternalCodeGenerator.physicalPartitionPlan(External.scala:126)
    at com.microsoft.dataflow.FlowRunner$$anonfun.apply(FlowRunner.scala:237)
    at com.microsoft.dataflow.FlowRunner$$anonfun.apply(FlowRunner.scala:216)
    at scala.collection.Iterator$$anon.next(Iterator.scala:410)
    at scala.collection.TraversableOnce$class.collectFirst(TraversableOnce.scala:145)
    at scala.collection.SeqViewLike$AbstractTransformed.collectFirst(SeqViewLike.scala:37)
    at com.microsoft.dataflow.FlowRunner$.com$microsoft$dataflow$FlowRunner$$runner(FlowRunner.scala:309)
    at com.microsoft.dataflow.FlowRunner$$anonfun$runner.apply(FlowRunner.scala:178)
    at com.microsoft.dataflow.FlowRunner$$anonfun$runner.apply(FlowRunner.scala:173)
    at scala.util.Success.flatMap(Try.scala:231)
    at com.microsoft.dataflow.FlowRunner$.runner(FlowRunner.scala:173)
    at com.microsoft.dataflow.DataflowExecutor$$anonfun$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun.apply(DataflowExecutor.scala:119)
    at com.microsoft.dataflow.DataflowExecutor$$anonfun$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun$apply$$anonfun.apply(DataflowExecutor.scala:106)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$flowCode.apply(DataflowJobFuture.scala:66)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$flowCode.apply(DataflowJobFuture.scala:66)
    at scala.Option.map(Option.scala:146)
    at com.microsoft.dataflow.DataflowJobFuture.flowCode$lzycompute(DataflowJobFuture.scala:66)
    at com.microsoft.dataflow.DataflowJobFuture.flowCode(DataflowJobFuture.scala:66)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply$mcV$sp(DataflowJobFuture.scala:290)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply(DataflowJobFuture.scala:287)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply(DataflowJobFuture.scala:287)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1(Future.scala:24)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply$mcV$sp(DataflowJobFuture.scala:315)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply(DataflowJobFuture.scala:287)
    at com.microsoft.dataflow.DataflowJobFuture$$anonfun$start.apply(DataflowJobFuture.scala:287)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1(Future.scala:24)
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

Answer 1

根据@Mark Kromer 的评论分享答案。不幸的是，“我们没有评估分区数量属性中的动态内容”。这被确认为错误，产品团队正在积极修复。

ADF 映射数据流 - Sink 转换动态分区数

ADF Mapping Data Flow - Sink transform dynamic Number of partitions

azure-data-factory

azure-data-factory-2