创建应用以运行管理命令

项目
10/02/2024

适用于：✅Azure 数据资源管理器

在本文中，学习如何：

运行管理命令并处理结果
更改表级别引入批处理策略
显示数据库级别保留策略

先决条件

设置开发环境以使用 Kusto 客户端库。

运行管理命令并处理结果

在首选 IDE 或文本编辑器中，使用适合首选语言的约定创建名为“管理命令”的项目或文件。然后，添加以下代码：

创建连接到群集的客户端应用。将 <your_cluster_uri> 占位符值替换为你的群集名。

注意

对于管理命令，你将使用 CreateCslAdminProvider 客户端工厂方法。

using Kusto.Data;
using Kusto.Data.Net.Client;

namespace ManagementCommands {
  class ManagementCommands {
    static void Main(string[] args) {
      var clusterUri = "<your_cluster_uri>";
      var kcsb = new KustoConnectionStringBuilder(clusterUri)
          .WithAadUserPromptAuthentication();

      using (var kustoClient = KustoClientFactory.CreateCslAdminProvider(kcsb)) {
      }
    }
  }
}

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder

def main():
  cluster_uri = "<your_cluster_uri>"
  kcsb = KustoConnectionStringBuilder.with_interactive_login(cluster_uri)

  with KustoClient(kcsb) as kusto_client:

if __name__ == "__main__":
  main()

import { Client as KustoClient, KustoConnectionStringBuilder } from "azure-kusto-data/";
import { InteractiveBrowserCredentialInBrowserOptions } from "@azure/identity";

async function main() {
  const clusterUri = "<your_cluster_uri>";
  const authOptions = {
    clientId: "5e39af3b-ba50-4255-b547-81abfb507c58",
    redirectUri: "http://localhost:5173",
  } as InteractiveBrowserCredentialInBrowserOptions;
  const kcsb = KustoConnectionStringBuilder.withUserPrompt(clusterUri, authOptions);
  const kustoClient = new KustoClient(kcsb);
}

main();

注意

对于 Node.js 应用，请使用 InteractiveBrowserCredentialNodeOptions 而不是 InteractiveBrowserCredentialInBrowserOptions。

import com.microsoft.azure.kusto.data.Client;
import com.microsoft.azure.kusto.data.ClientFactory;
import com.microsoft.azure.kusto.data.KustoOperationResult;
import com.microsoft.azure.kusto.data.KustoResultSetTable;
import com.microsoft.azure.kusto.data.KustoResultColumn;
import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;

public class ManagementCommands {
  public static void main(String[] args) throws Exception {
    try {
      String clusterUri = "<your_cluster_uri>";
      ConnectionStringBuilder kcsb = ConnectionStringBuilder.createWithUserPrompt(clusterUri);

      try (Client kustoClient = ClientFactory.createClient(kcsb)) {
      }
    }
  }
}

定义一个函数来打印正在运行的命令及其生成的表。此函数会解压缩结果表中的列名，并在新行中打印每个名称/值对。

static void PrintResultsAsValueList(string command, IDataReader response) {
  while (response.Read()) {
    Console.WriteLine("\n{0}\n", new String('-', 20));
    Console.WriteLine("Command: {0}", command);
    Console.WriteLine("Result:");
    for (int i = 0; i < response.FieldCount; i++) {
      Console.WriteLine("\t{0} - {1}", response.GetName(i), response.IsDBNull(i) ? "None" : response.GetString(i));
    }
  }
}

def print_result_as_value_list(command, response):
  # create a list of columns
  cols = (col.column_name for col in response.primary_results[0].columns)

  print("\n" + "-" * 20 + "\n")
  print("Command: " + command)
  # print the values for each row
  for row in response.primary_results[0]:
    print("Result:")
    for col in cols:
      print("\t", col, "-", row[col])

function printResultsAsValueList(command: string, response: KustoResponseDataSet) {
  // create a list of columns
  const cols = response.primaryResults[0].columns;

  console.log("\n" + "-".repeat(20) + "\n")
  console.log("Command: " + command)
  // print the values for each row
  for (const row of response.primaryResults[0].rows()) {
    console.log("Result:")
    for (col of cols)
    console.log("\t", col.name, "-", row.getValueAt(col.ordinal) ? row.getValueAt(col.ordinal).toString() : "None")
  }
}

public static void printResultsAsValueList(String command, KustoResultSetTable results) {
  while (results.next()) {
    System.out.println("\n" + "-".repeat(20) + "\n");
    System.out.println("Command: " + command);
    System.out.println("Result:");
    KustoResultColumn[] columns = results.getColumns();
    for (int i = 0; i < columns.length; i++) {
      System.out.println("\t" + columns[i].getColumnName() + " - " + (results.getObject(i) == null ? "None" : results.getString(i)));
    }
  }
}

定义要运行的命令。该命令会创建一个名为 MyStormEvents 的表，并将表架构定义为一个包含列名和类型的列表。将 <your_database> 占位符替换为你的数据库名。

string database = "<your_database>";
string table = "MyStormEvents";

// Create a table named MyStormEvents
// The brackets contain a list of column Name:Type pairs that defines the table schema
string command = @$".create table {table}
                  (StartTime:datetime,
                   EndTime:datetime,
                   State:string,
                   DamageProperty:int,
                   DamageCrops:int,
                   Source:string,
                   StormSummary:dynamic)";

database = "<your_database>"
table = "MyStormEvents"

# Create a table named MyStormEvents
# The brackets contain a list of column Name:Type pairs that defines the table schema
command = ".create table " + table + " " \
          "(StartTime:datetime," \
          " EndTime:datetime," \
          " State:string," \
          " DamageProperty:int," \
          " DamageCrops:int," \
          " Source:string," \
          " StormSummary:dynamic)"

const database = "<your_database>";
const table = "MyStormEvents";

// Create a table named MyStormEvents
// The brackets contain a list of column Name:Type pairs that defines the table schema
const command = `.create table ${table}
              (StartTime:datetime,
               EndTime:datetime,
               State:string,
               DamageProperty:int,
               Source:string,
               StormSummary:dynamic)`;

String database = "<your_database>";
String table = "MyStormEvents";

// Create a table named MyStormEvents
// The brackets contain a list of column Name:Type pairs that defines the table schema
String command = ".create table " + table + " " +
                 "(StartTime:datetime," +
                 " EndTime:datetime," +
                 " State:string," +
                 " DamageProperty:int," +
                 " DamageCrops:int," +
                 " Source:string," +
                 " StormSummary:dynamic)";

运行命令并使用之前定义的函数打印结果。

注意

你将使用 ExecuteControlCommand 方法来运行命令。

using (var response = kustoClient.ExecuteControlCommand(database, command, null)) {
  PrintResultsAsValueList(command, response);
}

注意

你将使用 execute_mgmt 方法来运行命令。

response = kusto_client.execute_mgmt(database, command)
print_result_as_value_list(command, response)

注意

你将使用 executeMgmt 方法来运行命令。

const response = await kustoClient.executeMgmt(database, command);
printResultsAsValueList(command, response)

KustoOperationResult response = kusto_client.execute(database, command);
printResultsAsValueList(command, response.getPrimaryResults());

完整代码应如下所示：

using Kusto.Data;
using Kusto.Data.Net.Client;

namespace ManagementCommands {
  class ManagementCommands {
    static void Main(string[] args) {
      string clusterUri = "https://<your_cluster_uri>";
      var kcsb = new KustoConnectionStringBuilder(clusterUri)
          .WithAadUserPromptAuthentication();

      using (var kustoClient = KustoClientFactory.CreateCslAdminProvider(kcsb)) {
        string database = "<your_database>";
        string table = "MyStormEvents";

        // Create a table named MyStormEvents
        // The brackets contain a list of column Name:Type pairs that defines the table schema
        string command = @$".create table {table} 
                          (StartTime:datetime,
                           EndTime:datetime,
                           State:string,
                           DamageProperty:int,
                           DamageCrops:int,
                           Source:string,
                           StormSummary:dynamic)";

        using (var response = kustoClient.ExecuteControlCommand(database, command, null)) {
          PrintResultsAsValueList(command, response);
        }
      }
    }

    static void PrintResultsAsValueList(string command, IDataReader response) {
      while (response.Read()) {
        Console.WriteLine("\n{0}\n", new String('-', 20));
        Console.WriteLine("Command: {0}", command);
        Console.WriteLine("Result:");
        for (int i = 0; i < response.FieldCount; i++) {
          Console.WriteLine("\t{0} - {1}", response.GetName(i), response.IsDBNull(i) ? "None" : response.GetString(i));
        }
      }
    }
  }
}

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder

def main():
  cluster_uri = "https://<your_cluster_uri>"
  kcsb = KustoConnectionStringBuilder.with_interactive_login(cluster_uri)

  with KustoClient(kcsb) as kusto_client:

    database = "<your_database>"
    table = "MyStormEvents"

    # Create a table named MyStormEvents
    # The brackets contain a list of column Name:Type pairs that defines the table schema
    command = ".create table " + table + " " \
              "(StartTime:datetime," \
              " EndTime:datetime," \
              " State:string," \
              " DamageProperty:int," \
              " DamageCrops:int," \
              " Source:string," \
              " StormSummary:dynamic)"

    response = kusto_client.execute_mgmt(database, command)
    print_result_as_value_list(command, response)

def print_result_as_value_list(command, response):
  # create a list of columns
  cols = (col.column_name for col in response.primary_results[0].columns)

  print("\n" + "-" * 20 + "\n")
  print("Command: " + command)
  # print the values for each row
  for row in response.primary_results[0]:
    print("Result:")
    for col in cols:
      print("\t", col, "-", row[col])

if __name__ == "__main__":
  main()

import { Client as KustoClient, KustoConnectionStringBuilder, KustoResponseDataSet } from "azure-kusto-data/";
import { InteractiveBrowserCredentialInBrowserOptions } from "@azure/identity";

async function main() {
  const clusterUri = "<your_cluster_uri>";
  const authOptions = {
    clientId: "5e39af3b-ba50-4255-b547-81abfb507c58",
    redirectUri: "http://localhost:5173",
  } as InteractiveBrowserCredentialInBrowserOptions;
  const kcsb = KustoConnectionStringBuilder.withUserPrompt(clusterUri, authOptions);
  const kustoClient = new KustoClient(kcsb);

  const database = "<your_database>";
  const table = "MyStormEvents";

  // Create a table named MyStormEvents
  // The brackets contain a list of column Name:Type pairs that defines the table schema
  const command = `.create table ${table}
                 (StartTime:datetime,
                  EndTime:datetime,
                  State:string,
                  DamageProperty:int,
                  Source:string,
                  StormSummary:dynamic)`;

  const response = await kustoClient.executeMgmt(database, command);
  printResultsAsValueList(command, response)
}

function printResultsAsValueList(command: string, response: KustoResponseDataSet) {
  // create a list of columns
  const cols = response.primaryResults[0].columns;

  console.log("\n" + "-".repeat(20) + "\n")
  console.log("Command: " + command)
  // print the values for each row
  for (const row of response.primaryResults[0].rows()) {
    console.log("Result:")
    for (const col of cols) {
        console.log("\t", col.name, "-", row.getValueAt(col.ordinal) ? row.getValueAt(col.ordinal).toString() : "None")
    }
  }
}

main();

注意

对于 Node.js 应用，请使用 InteractiveBrowserCredentialNodeOptions 而不是 InteractiveBrowserCredentialInBrowserOptions。

import com.microsoft.azure.kusto.data.Client;
import com.microsoft.azure.kusto.data.ClientFactory;
import com.microsoft.azure.kusto.data.KustoOperationResult;
import com.microsoft.azure.kusto.data.KustoResultSetTable;
import com.microsoft.azure.kusto.data.KustoResultColumn;
import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;

public class ManagementCommands {
  public static void main(String[] args) throws Exception {
    try {
      String clusterUri = "https://<your_cluster_uri>";
      ConnectionStringBuilder kcsb = ConnectionStringBuilder.createWithUserPrompt(clusterUri);
      try (Client kustoClient = ClientFactory.createClient(kcsb)) {
        String database = "<your_database>";
        String table = "MyStormEvents";

        // Create a table named MyStormEvents
        // The brackets contain a list of column Name:Type pairs that defines the table schema
        String command = ".create table " + table + " " +
                         "(StartTime:datetime," +
                         " EndTime:datetime," +
                         " State:string," +
                         " DamageProperty:int," +
                         " DamageCrops:int," +
                         " Source:string," +
                         " StormSummary:dynamic)";

        KustoOperationResult response = kustoClient.execute(database, command);
        printResultsAsValueList(command, response.getPrimaryResults());
      }
    }
  }
  public static void printResultsAsValueList(String command, KustoResultSetTable results) {
    while (results.next()) {
      System.out.println("\n" + "-".repeat(20) + "\n");
      System.out.println("Command: " + command);
      System.out.println("Result:");
      KustoResultColumn[] columns = results.getColumns();
      for (int i = 0; i < columns.length; i++) {
        System.out.println("\t" + columns[i].getColumnName() + " - " + (results.getObject(i) == null ? "None" : results.getString(i)));
      }
    }
  }
}

运行应用

在命令行界面中，使用以下命令运行应用：

# Change directory to the folder that contains the management commands project
dotnet run .

python management_commands.py

在 Node.js 环境中：

node management-commands.js

在浏览器环境中，使用适当的命令来运行应用。例如，对于 Vite-React，请使用以下命令：

npm run dev

注意

在浏览器环境中，打开开发人员工具控制台以查看输出。

mvn install exec:java -Dexec.mainClass="<groupId>.ManagementCommands"

你应会看到如下所示的结果：

--------------------

Command: .create table MyStormEvents 
                 (StartTime:datetime,
                  EndTime:datetime,
                  State:string,
                  DamageProperty:int,
                  Source:string,
                  StormSummary:dynamic)
Result:
   TableName - MyStormEvents
   Schema - {"Name":"MyStormEvents","OrderedColumns":[{"Name":"StartTime","Type":"System.DateTime","CslType":"datetime"},{"Name":"EndTime","Type":"System.DateTime","CslType":"datetime"},{"Name":"State","Type":"System.String","CslType":"string"},{"Name":"DamageProperty","Type":"System.Int32","CslType":"int"},{"Name":"Source","Type":"System.String","CslType":"string"},{"Name":"StormSummary","Type":"System.Object","CslType":"dynamic"}]}
   DatabaseName - MyDatabaseName
   Folder - None
   DocString - None

更改表级别引入批处理策略

可通过更改相应的表策略来自定义表的引入批处理行为。有关详细信息，请参阅引入批处理策略。

注意

如果未指定 PolicyObject 的所有参数，则未指定的参数将设置为默认值。例如，仅指定“MaximumBatchingTimeSpan”会导致将“MaximumNumberOfItems”和“MaximumRawDataSizeMB”设置为默认值。

例如，可使用以下命令更改 MyStormEvents 表的 ingestionBatching 策略，从而修改应用以将引入批处理策略超时值更改为 30 秒：

// Reduce the default batching timeout to 30 seconds
command = @$".alter-merge table {table} policy ingestionbatching '{{ ""MaximumBatchingTimeSpan"":""00:00:30"" }}'";

using (var response = kustoClient.ExecuteControlCommand(database, command, null))
{
  PrintResultsAsValueList(command, response);
}

# Reduce the default batching timeout to 30 seconds
command = ".alter-merge table " + table + " policy ingestionbatching '{ \"MaximumBatchingTimeSpan\":\"00:00:30\" }'"

response = kusto_client.execute_mgmt(database, command)
print_result_as_value_list(command, response)

// Reduce the default batching timeout to 30 seconds
command = ".alter-merge table " + table + " policy ingestionbatching '{ \"MaximumBatchingTimeSpan\":\"00:00:30\" }'"

response = await kustoClient.executeMgmt(database, command)
printResultsAsValueList(command, response)

// Reduce the default batching timeout to 30 seconds
command = ".alter-merge table " + table + " policy ingestionbatching '{ \"MaximumBatchingTimeSpan\":\"00:00:30\" }'";

response = kusto_client.execute(database, command);
printResultsAsValueList(command, response.getPrimaryResults());

将代码添加到应用并运行它时，应该会看到如下所示的结果：

--------------------

Command: .alter-merge table MyStormEvents policy ingestionbatching '{ "MaximumBatchingTimeSpan":"00:00:30" }'
Result:
   PolicyName - IngestionBatchingPolicy
   EntityName - [YourDatabase].[MyStormEvents]
   Policy - {
  "MaximumBatchingTimeSpan": "00:00:30",
  "MaximumNumberOfItems": 500,
  "MaximumRawDataSizeMB": 1024
}
   ChildEntities - None
   EntityType - Table

显示数据库级别保留策略

可使用管理命令来显示数据库的保留策略。

例如，可使用以下代码修改应用来显示数据库的保留策略。对结果进行策展，投射出结果中的两列内容：

// Show the database retention policy (drop some columns from the result)
command = @$".show database {database} policy retention | project-away ChildEntities, EntityType";

using (var response = kustoClient.ExecuteControlCommand(database, command, null)) {
  PrintResultsAsValueList(command, response);
}

# Show the database retention policy (drop some columns from the result)
command = ".show database " + database + " policy retention | project-away ChildEntities, EntityType"

response = kusto_client.execute_mgmt(database, command)
print_result_as_value_list(command, response)

// Show the database retention policy (drop some columns from the result)
command = ".show database " + database + " policy retention | project-away ChildEntities, EntityType"

response = await kustoClient.executeMgmt(database, command)
printResultsAsValueList(command, response)

// Show the database retention policy (drop some columns from the result)
command = ".show database " + database + " policy retention | project-away ChildEntities, EntityType";

response = kusto_client.execute(database, command);
printResultsAsValueList(command, response.getPrimaryResults());