Modules

Modules allow users to extend Flink’s built-in objects, such as defining functions that behave like Flink built-in functions. They are pluggable, and while Flink provides a few pre-built modules, users can write their own.

For example, users can define their own geo functions and plug them into Flink as built-in functions to be used in Flink SQL and Table APIs. Another example is users can load an out-of-shelf Hive module to use Hive built-in functions as Flink built-in functions.

Furthermore, a module can provide built-in table source and sink factories which disable Flink’s default discovery mechanism based on Java’s Service Provider Interfaces (SPI), or influence how connectors of temporary tables should be created without a corresponding catalog.

Module Types

CoreModule

CoreModule contains all of Flink’s system (built-in) functions and is loaded and enabled by default.

HiveModule

The HiveModule provides Hive built-in functions as Flink’s system functions to SQL and Table API users. Flink’s Hive documentation provides full details on setting up the module.

User-Defined Module

Users can develop custom modules by implementing the Module interface. To use custom modules in SQL CLI, users should develop both a module and its corresponding module factory by implementing the ModuleFactory interface.

A module factory defines a set of properties for configuring the module when the SQL CLI bootstraps. Properties are passed to a discovery service where the service tries to match the properties to a ModuleFactory and instantiate a corresponding module instance.

Module Lifecycle and Resolution Order

A module can be loaded, enabled, disabled and unloaded. When TableEnvironment loads a module initially, it enables the module by default. Flink supports multiple modules and keeps track of the loading order to resolve metadata. Besides, Flink only resolves the functions among enabled modules. E.g., when there are two functions of the same name residing in two modules, there will be three conditions.

  • If both of the modules are enabled, then Flink resolves the function according to the resolution order of the modules.
  • If one of them is disabled, then Flink resolves the function to the enabled module.
  • If both of the modules are disabled, then Flink cannot resolve the function.

Users can change the resolution order by using modules in a different declared order. E.g., users can specify Flink to find functions first in Hive by USE MODULES hive, core.

Besides, users can also disable modules by not declaring them. E.g., users can specify Flink to disable core module by USE MODULES hive (However, it is strongly not recommended disabling core module). Disable a module does not unload it, and users can enable it again by using it. E.g., users can bring back core module and place it in the first by USE MODULES core, hive. A module can be enabled only when it is loaded already. Using an unloaded module will throw an Exception. Eventually, users can unload a module.

The difference between disabling and unloading a module is that TableEnvironment still keeps the disabled modules, and users can list all loaded modules to view the disabled modules.

Namespace

Objects provided by modules are considered part of Flink’s system (built-in) objects; thus, they don’t have any namespaces.

How to Load, Unload, Use and List Modules

Using SQL

Users can use SQL to load/unload/use/list modules in both Table API and SQL CLI.

Java

  1. EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
  2. TableEnvironment tableEnv = TableEnvironment.create(settings);
  4. // Show initially loaded and enabled modules
  5. tableEnv.executeSql("SHOW MODULES").print();
  6. // +-------------+
  7. // | module name |
  8. // +-------------+
  9. // |        core |
  10. // +-------------+
  11. tableEnv.executeSql("SHOW FULL MODULES").print();
  12. // +-------------+------+
  13. // | module name | used |
  14. // +-------------+------+
  15. // |        core | true |
  16. // +-------------+------+
  18. // Load a hive module
  19. tableEnv.executeSql("LOAD MODULE hive WITH ('hive-version' = '...')");
  21. // Show all enabled modules
  22. tableEnv.executeSql("SHOW MODULES").print();
  23. // +-------------+
  24. // | module name |
  25. // +-------------+
  26. // |        core |
  27. // |        hive |
  28. // +-------------+
  30. // Show all loaded modules with both name and use status
  31. tableEnv.executeSql("SHOW FULL MODULES").print();
  32. // +-------------+------+
  33. // | module name | used |
  34. // +-------------+------+
  35. // |        core | true |
  36. // |        hive | true |
  37. // +-------------+------+
  39. // Change resolution order
  40. tableEnv.executeSql("USE MODULES hive, core");
  41. tableEnv.executeSql("SHOW MODULES").print();
  42. // +-------------+
  43. // | module name |
  44. // +-------------+
  45. // |        hive |
  46. // |        core |
  47. // +-------------+
  48. tableEnv.executeSql("SHOW FULL MODULES").print();
  49. // +-------------+------+
  50. // | module name | used |
  51. // +-------------+------+
  52. // |        hive | true |
  53. // |        core | true |
  54. // +-------------+------+
  56. // Disable core module
  57. tableEnv.executeSql("USE MODULES hive");
  58. tableEnv.executeSql("SHOW MODULES").print();
  59. // +-------------+
  60. // | module name |
  61. // +-------------+
  62. // |        hive |
  63. // +-------------+
  64. tableEnv.executeSql("SHOW FULL MODULES").print();
  65. // +-------------+-------+
  66. // | module name |  used |
  67. // +-------------+-------+
  68. // |        hive |  true |
  69. // |        core | false |
  70. // +-------------+-------+
  72. // Unload hive module
  73. tableEnv.executeSql("UNLOAD MODULE hive");
  74. tableEnv.executeSql("SHOW MODULES").print();
  75. // Empty set
  76. tableEnv.executeSql("SHOW FULL MODULES").print();
  77. // +-------------+-------+
  78. // | module name |  used |
  79. // +-------------+-------+
  80. // |        hive | false |
  81. // +-------------+-------+

Scala

  1. val settings = EnvironmentSettings.inStreamingMode()
  2. val tableEnv = TableEnvironment.create(setting)
  4. // Show initially loaded and enabled modules
  5. tableEnv.executeSql("SHOW MODULES").print()
  6. // +-------------+
  7. // | module name |
  8. // +-------------+
  9. // |        core |
  10. // +-------------+
  11. tableEnv.executeSql("SHOW FULL MODULES").print()
  12. // +-------------+------+
  13. // | module name | used |
  14. // +-------------+------+
  15. // |        core | true |
  16. // +-------------+------+
  18. // Load a hive module
  19. tableEnv.executeSql("LOAD MODULE hive WITH ('hive-version' = '...')")
  21. // Show all enabled modules
  22. tableEnv.executeSql("SHOW MODULES").print()
  23. // +-------------+
  24. // | module name |
  25. // +-------------+
  26. // |        core |
  27. // |        hive |
  28. // +-------------+
  30. // Show all loaded modules with both name and use status
  31. tableEnv.executeSql("SHOW FULL MODULES")
  32. // +-------------+------+
  33. // | module name | used |
  34. // +-------------+------+
  35. // |        core | true |
  36. // |        hive | true |
  37. // +-------------+------+
  39. // Change resolution order
  40. tableEnv.executeSql("USE MODULES hive, core")
  41. tableEnv.executeSql("SHOW MODULES").print()
  42. // +-------------+
  43. // | module name |
  44. // +-------------+
  45. // |        hive |
  46. // |        core |
  47. // +-------------+
  48. tableEnv.executeSql("SHOW FULL MODULES").print()
  49. // +-------------+------+
  50. // | module name | used |
  51. // +-------------+------+
  52. // |        hive | true |
  53. // |        core | true |
  54. // +-------------+------+
  56. // Disable core module
  57. tableEnv.executeSql("USE MODULES hive")
  58. tableEnv.executeSql("SHOW MODULES").print()
  59. // +-------------+
  60. // | module name |
  61. // +-------------+
  62. // |        hive |
  63. // +-------------+
  64. tableEnv.executeSql("SHOW FULL MODULES").print()
  65. // +-------------+-------+
  66. // | module name |  used |
  67. // +-------------+-------+
  68. // |        hive |  true |
  69. // |        core | false |
  70. // +-------------+-------+
  72. // Unload hive module
  73. tableEnv.executeSql("UNLOAD MODULE hive")
  74. tableEnv.executeSql("SHOW MODULES").print()
  75. // Empty set
  76. tableEnv.executeSql("SHOW FULL MODULES").print()
  77. // +-------------+-------+
  78. // | module name |  used |
  79. // +-------------+-------+
  80. // |        hive | false |
  81. // +-------------+-------+

Python

  1. from pyflink.table import *
  3. # environment configuration
  4. settings = EnvironmentSettings.inStreamingMode()
  5. t_env = TableEnvironment.create(settings)
  7. # Show initially loaded and enabled modules
  8. t_env.execute_sql("SHOW MODULES").print()
  9. # +-------------+
  10. # | module name |
  11. # +-------------+
  12. # |        core |
  13. # +-------------+
  14. t_env.execute_sql("SHOW FULL MODULES").print()
  15. # +-------------+------+
  16. # | module name | used |
  17. # +-------------+------+
  18. # |        core | true |
  19. # +-------------+------+
  21. # Load a hive module
  22. t_env.execute_sql("LOAD MODULE hive WITH ('hive-version' = '...')")
  24. # Show all enabled modules
  25. t_env.execute_sql("SHOW MODULES").print()
  26. # +-------------+
  27. # | module name |
  28. # +-------------+
  29. # |        core |
  30. # |        hive |
  31. # +-------------+
  33. # Show all loaded modules with both name and use status
  34. t_env.execute_sql("SHOW FULL MODULES").print()
  35. # +-------------+------+
  36. # | module name | used |
  37. # +-------------+------+
  38. # |        core | true |
  39. # |        hive | true |
  40. # +-------------+------+
  42. # Change resolution order
  43. t_env.execute_sql("USE MODULES hive, core")
  44. t_env.execute_sql("SHOW MODULES").print()
  45. # +-------------+
  46. # | module name |
  47. # +-------------+
  48. # |        hive |
  49. # |        core |
  50. # +-------------+
  51. t_env.execute_sql("SHOW FULL MODULES").print()
  52. # +-------------+------+
  53. # | module name | used |
  54. # +-------------+------+
  55. # |        hive | true |
  56. # |        core | true |
  57. # +-------------+------+
  59. # Disable core module
  60. t_env.execute_sql("USE MODULES hive")
  61. t_env.execute_sql("SHOW MODULES").print()
  62. # +-------------+
  63. # | module name |
  64. # +-------------+
  65. # |        hive |
  66. # +-------------+
  67. t_env.execute_sql("SHOW FULL MODULES").print()
  68. # +-------------+-------+
  69. # | module name |  used |
  70. # +-------------+-------+
  71. # |        hive |  true |
  72. # |        core | false |
  73. # +-------------+-------+
  75. # Unload hive module
  76. t_env.execute_sql("UNLOAD MODULE hive")
  77. t_env.execute_sql("SHOW MODULES").print()
  78. # Empty set
  79. t_env.execute_sql("SHOW FULL MODULES").print()
  80. # +-------------+-------+
  81. # | module name |  used |
  82. # +-------------+-------+
  83. # |        hive | false |
  84. # +-------------+-------+

SQL Client

  1. -- Show initially loaded and enabled modules
  2. Flink SQL> SHOW MODULES;
  3. +-------------+
  4. | module name |
  5. +-------------+
  6. |        core |
  7. +-------------+
  8. 1 row in set
  9. Flink SQL> SHOW FULL MODULES;
  10. +-------------+------+
  11. | module name | used |
  12. +-------------+------+
  13. |        core | true |
  14. +-------------+------+
  15. 1 row in set
  17. -- Load a hive module
  18. Flink SQL> LOAD MODULE hive WITH ('hive-version' = '...');
  20. -- Show all enabled modules
  21. Flink SQL> SHOW MODULES;
  22. +-------------+
  23. | module name |
  24. +-------------+
  25. |        core |
  26. |        hive |
  27. +-------------+
  28. 2 rows in set
  30. -- Show all loaded modules with both name and use status
  31. Flink SQL> SHOW FULL MODULES;
  32. +-------------+------+
  33. | module name | used |
  34. +-------------+------+
  35. |        core | true |
  36. |        hive | true |
  37. +-------------+------+
  38. 2 rows in set
  40. -- Change resolution order
  41. Flink SQL> USE MODULES hive, core ;
  42. Flink SQL> SHOW MODULES;
  43. +-------------+
  44. | module name |
  45. +-------------+
  46. |        hive |
  47. |        core |
  48. +-------------+
  49. 2 rows in set
  50. Flink SQL> SHOW FULL MODULES;
  51. +-------------+------+
  52. | module name | used |
  53. +-------------+------+
  54. |        hive | true |
  55. |        core | true |
  56. +-------------+------+
  57. 2 rows in set
  59. -- Unload hive module
  60. Flink SQL> UNLOAD MODULE hive;
  61. Flink SQL> SHOW MODULES;
  62. Empty set
  63. Flink SQL> SHOW FULL MODULES;
  64. +-------------+-------+
  65. | module name |  used |
  66. +-------------+-------+
  67. |        hive | false |
  68. +-------------+-------+
  69. 1 row in set

YAML

All modules defined using YAML must provide a type property that specifies the type. The following types are supported out of the box.

ModuleType Value
CoreModulecore
HiveModulehive
  1. modules:
  2.    - name: core
  3.      type: core
  4.    - name: hive
  5.      type: hive

When using SQL, module name is used to perform the module discovery. It is parsed as a simple identifier and case-sensitive.

Using Java, Scala or Python

Users can use Java, Scala or Python to load/unload/use/list modules programmatically.

Java

  1. EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
  2. TableEnvironment tableEnv = TableEnvironment.create(settings);
  4. // Show initially loaded and enabled modules
  5. tableEnv.listModules();
  6. // +-------------+
  7. // | module name |
  8. // +-------------+
  9. // |        core |
  10. // +-------------+
  11. tableEnv.listFullModules();
  12. // +-------------+------+
  13. // | module name | used |
  14. // +-------------+------+
  15. // |        core | true |
  16. // +-------------+------+
  18. // Load a hive module
  19. tableEnv.loadModule("hive", new HiveModule());
  21. // Show all enabled modules
  22. tableEnv.listModules();
  23. // +-------------+
  24. // | module name |
  25. // +-------------+
  26. // |        core |
  27. // |        hive |
  28. // +-------------+
  30. // Show all loaded modules with both name and use status
  31. tableEnv.listFullModules();
  32. // +-------------+------+
  33. // | module name | used |
  34. // +-------------+------+
  35. // |        core | true |
  36. // |        hive | true |
  37. // +-------------+------+
  39. // Change resolution order
  40. tableEnv.useModules("hive", "core");
  41. tableEnv.listModules();
  42. // +-------------+
  43. // | module name |
  44. // +-------------+
  45. // |        hive |
  46. // |        core |
  47. // +-------------+
  48. tableEnv.listFullModules();
  49. // +-------------+------+
  50. // | module name | used |
  51. // +-------------+------+
  52. // |        hive | true |
  53. // |        core | true |
  54. // +-------------+------+
  56. // Disable core module
  57. tableEnv.useModules("hive");
  58. tableEnv.listModules();
  59. // +-------------+
  60. // | module name |
  61. // +-------------+
  62. // |        hive |
  63. // +-------------+
  64. tableEnv.listFullModules();
  65. // +-------------+-------+
  66. // | module name |  used |
  67. // +-------------+-------+
  68. // |        hive |  true |
  69. // |        core | false |
  70. // +-------------+-------+
  72. // Unload hive module
  73. tableEnv.unloadModule("hive");
  74. tableEnv.listModules();
  75. // Empty set
  76. tableEnv.listFullModules();
  77. // +-------------+-------+
  78. // | module name |  used |
  79. // +-------------+-------+
  80. // |        hive | false |
  81. // +-------------+-------+

Scala

  1. val settings = EnvironmentSettings.inStreamingMode()
  2. val tableEnv = TableEnvironment.create(setting)
  4. // Show initially loaded and enabled modules
  5. tableEnv.listModules()
  6. // +-------------+
  7. // | module name |
  8. // +-------------+
  9. // |        core |
  10. // +-------------+
  11. tableEnv.listFullModules()
  12. // +-------------+------+
  13. // | module name | used |
  14. // +-------------+------+
  15. // |        core | true |
  16. // +-------------+------+
  18. // Load a hive module
  19. tableEnv.loadModule("hive", new HiveModule())
  21. // Show all enabled modules
  22. tableEnv.listModules()
  23. // +-------------+
  24. // | module name |
  25. // +-------------+
  26. // |        core |
  27. // |        hive |
  28. // +-------------+
  30. // Show all loaded modules with both name and use status
  31. tableEnv.listFullModules()
  32. // +-------------+------+
  33. // | module name | used |
  34. // +-------------+------+
  35. // |        core | true |
  36. // |        hive | true |
  37. // +-------------+------+
  39. // Change resolution order
  40. tableEnv.useModules("hive", "core")
  41. tableEnv.listModules()
  42. // +-------------+
  43. // | module name |
  44. // +-------------+
  45. // |        hive |
  46. // |        core |
  47. // +-------------+
  48. tableEnv.listFullModules()
  49. // +-------------+------+
  50. // | module name | used |
  51. // +-------------+------+
  52. // |        hive | true |
  53. // |        core | true |
  54. // +-------------+------+
  56. // Disable core module
  57. tableEnv.useModules("hive")
  58. tableEnv.listModules()
  59. // +-------------+
  60. // | module name |
  61. // +-------------+
  62. // |        hive |
  63. // +-------------+
  64. tableEnv.listFullModules()
  65. // +-------------+-------+
  66. // | module name |  used |
  67. // +-------------+-------+
  68. // |        hive |  true |
  69. // |        core | false |
  70. // +-------------+-------+
  72. // Unload hive module
  73. tableEnv.unloadModule("hive")
  74. tableEnv.listModules()
  75. // Empty set
  76. tableEnv.listFullModules()
  77. // +-------------+-------+
  78. // | module name |  used |
  79. // +-------------+-------+
  80. // |        hive | false |
  81. // +-------------+-------+

Python

  1. from pyflink.table import *
  3. # environment configuration
  4. settings = EnvironmentSettings.inStreamingMode()
  5. t_env = TableEnvironment.create(settings)
  7. # Show initially loaded and enabled modules
  8. t_env.list_modules()
  9. # +-------------+
  10. # | module name |
  11. # +-------------+
  12. # |        core |
  13. # +-------------+
  14. t_env.list_full_modules()
  15. # +-------------+------+
  16. # | module name | used |
  17. # +-------------+------+
  18. # |        core | true |
  19. # +-------------+------+
  21. # Load a hive module
  22. t_env.load_module("hive", HiveModule())
  24. # Show all enabled modules
  25. t_env.list_modules()
  26. # +-------------+
  27. # | module name |
  28. # +-------------+
  29. # |        core |
  30. # |        hive |
  31. # +-------------+
  33. # Show all loaded modules with both name and use status
  34. t_env.list_full_modules()
  35. # +-------------+------+
  36. # | module name | used |
  37. # +-------------+------+
  38. # |        core | true |
  39. # |        hive | true |
  40. # +-------------+------+
  42. # Change resolution order
  43. t_env.use_modules("hive", "core")
  44. t_env.list_modules()
  45. # +-------------+
  46. # | module name |
  47. # +-------------+
  48. # |        hive |
  49. # |        core |
  50. # +-------------+
  51. t_env.list_full_modules()
  52. # +-------------+------+
  53. # | module name | used |
  54. # +-------------+------+
  55. # |        hive | true |
  56. # |        core | true |
  57. # +-------------+------+
  59. # Disable core module
  60. t_env.use_modules("hive")
  61. t_env.list_modules()
  62. # +-------------+
  63. # | module name |
  64. # +-------------+
  65. # |        hive |
  66. # +-------------+
  67. t_env.list_full_modules()
  68. # +-------------+-------+
  69. # | module name |  used |
  70. # +-------------+-------+
  71. # |        hive |  true |
  72. # |        core | false |
  73. # +-------------+-------+
  75. # Unload hive module
  76. t_env.unload_module("hive")
  77. t_env.list_modules()
  78. # Empty set
  79. t_env.list_full_modules()
  80. # +-------------+-------+
  81. # | module name |  used |
  82. # +-------------+-------+
  83. # |        hive | false |
  84. # +-------------+-------+